Skip to content

Oauth2

Module dedicated to OAuth2 device flow.

DeviceGrant

Bases: GrantMethodBase

Device grant method.

Source code in teledetection/sdk/oauth2.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class DeviceGrant(GrantMethodBase):
    """Device grant method."""

    client_id = "gdal"

    def get_first_token(self) -> JWT:
        """Get the first JWT token."""
        device_endpoint = f"{self.get_token_endpoint().rsplit('/', 1)[0]}/auth/device"

        req = create_session()
        log.debug("Getting token using device authorization grant")
        ret = req.post(
            device_endpoint,
            headers=self.headers,
            data=self.data_base,
            timeout=TIMEOUT,
        )
        if ret.status_code == 200:
            response = DeviceGrantResponse(**ret.json())
            verif_url_comp = response.verification_uri_complete
            log.info("Open the following URL in your browser to grant access:")
            log.info("\033[92m %s \033[0m", verif_url_comp)

            # QR code
            qr_code = qrcode.QRCode()
            qr_code.add_data(verif_url_comp)
            buffer = io.StringIO()
            qr_code.print_ascii(out=buffer)
            buffer.seek(0)
            log.info(buffer.read())

            log.info("Waiting for authentication...")
            start = time.time()
            data = self.data_base.copy()
            grant_type = "urn:ietf:params:oauth:grant-type:device_code"
            data.update(
                {
                    "device_code": response.device_code,
                    "grant_type": grant_type,
                }
            )
            while True:
                ret = req.post(
                    self.get_token_endpoint(),
                    headers=self.headers,
                    data=data,
                    timeout=TIMEOUT,
                )
                elapsed = time.time() - start
                log.debug(
                    "Elapsed: %.0f, status: %s (%.0f seconds left)",
                    elapsed,
                    ret.status_code,
                    response.expires_in - elapsed,
                )
                if elapsed > response.expires_in:
                    raise ExpiredAuthLinkError  # pragma: no cover
                if ret.status_code != 200:
                    time.sleep(response.interval)
                else:
                    return JWT.from_dict(ret.json())
        raise ConnectionError("Unable to authenticate with the SSO")

get_first_token()

Get the first JWT token.

Source code in teledetection/sdk/oauth2.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def get_first_token(self) -> JWT:
    """Get the first JWT token."""
    device_endpoint = f"{self.get_token_endpoint().rsplit('/', 1)[0]}/auth/device"

    req = create_session()
    log.debug("Getting token using device authorization grant")
    ret = req.post(
        device_endpoint,
        headers=self.headers,
        data=self.data_base,
        timeout=TIMEOUT,
    )
    if ret.status_code == 200:
        response = DeviceGrantResponse(**ret.json())
        verif_url_comp = response.verification_uri_complete
        log.info("Open the following URL in your browser to grant access:")
        log.info("\033[92m %s \033[0m", verif_url_comp)

        # QR code
        qr_code = qrcode.QRCode()
        qr_code.add_data(verif_url_comp)
        buffer = io.StringIO()
        qr_code.print_ascii(out=buffer)
        buffer.seek(0)
        log.info(buffer.read())

        log.info("Waiting for authentication...")
        start = time.time()
        data = self.data_base.copy()
        grant_type = "urn:ietf:params:oauth:grant-type:device_code"
        data.update(
            {
                "device_code": response.device_code,
                "grant_type": grant_type,
            }
        )
        while True:
            ret = req.post(
                self.get_token_endpoint(),
                headers=self.headers,
                data=data,
                timeout=TIMEOUT,
            )
            elapsed = time.time() - start
            log.debug(
                "Elapsed: %.0f, status: %s (%.0f seconds left)",
                elapsed,
                ret.status_code,
                response.expires_in - elapsed,
            )
            if elapsed > response.expires_in:
                raise ExpiredAuthLinkError  # pragma: no cover
            if ret.status_code != 200:
                time.sleep(response.interval)
            else:
                return JWT.from_dict(ret.json())
    raise ConnectionError("Unable to authenticate with the SSO")

ExpiredAuthLinkError

Bases: Exception

Authentication link has expired.

Source code in teledetection/sdk/oauth2.py 
37
38
class ExpiredAuthLinkError(Exception):
    """Authentication link has expired."""

GrantMethodBase

Base class for grant methods.

Source code in teledetection/sdk/oauth2.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class GrantMethodBase:
    """Base class for grant methods."""

    client_id: str
    headers: Dict[str, str] = {"Content-Type": "application/x-www-form-urlencoded"}
    _token_endpoint: str | None = None

    def get_token_endpoint(self):
        """Get the token endpoint."""
        if not self._token_endpoint:
            self._token_endpoint = retrieve_token_endpoint()
        return retrieve_token_endpoint()

    @abstractmethod
    def get_first_token(self) -> JWT:
        """Provide the first used token."""
        raise NotImplementedError

    @property
    def data_base(self) -> Dict[str, str]:
        """Base payload."""
        return {
            "client_id": self.client_id,
            "scope": "openid offline_access",
        }

    def refresh_token(self, old_jwt: JWT) -> JWT:
        """Refresh the token."""
        log.debug("Refreshing token")
        assert old_jwt, "JWT is empty"
        data = self.data_base.copy()
        data.update(
            {
                "refresh_token": old_jwt.refresh_token,
                "grant_type": "refresh_token",
            }
        )
        ret = create_session().post(
            self.get_token_endpoint(),
            headers=self.headers,
            data=data,
            timeout=TIMEOUT,
        )
        if ret.status_code == 200:
            log.debug(ret.text)
            return JWT.from_dict(ret.json())
        raise RefreshTokenError

data_base property

Base payload.

get_first_token() abstractmethod

Provide the first used token.

Source code in teledetection/sdk/oauth2.py 
54
55
56
57
@abstractmethod
def get_first_token(self) -> JWT:
    """Provide the first used token."""
    raise NotImplementedError

get_token_endpoint()

Get the token endpoint.

Source code in teledetection/sdk/oauth2.py 
48
49
50
51
52
def get_token_endpoint(self):
    """Get the token endpoint."""
    if not self._token_endpoint:
        self._token_endpoint = retrieve_token_endpoint()
    return retrieve_token_endpoint()

refresh_token(old_jwt)

Refresh the token.

Source code in teledetection/sdk/oauth2.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def refresh_token(self, old_jwt: JWT) -> JWT:
    """Refresh the token."""
    log.debug("Refreshing token")
    assert old_jwt, "JWT is empty"
    data = self.data_base.copy()
    data.update(
        {
            "refresh_token": old_jwt.refresh_token,
            "grant_type": "refresh_token",
        }
    )
    ret = create_session().post(
        self.get_token_endpoint(),
        headers=self.headers,
        data=data,
        timeout=TIMEOUT,
    )
    if ret.status_code == 200:
        log.debug(ret.text)
        return JWT.from_dict(ret.json())
    raise RefreshTokenError

OAuth2Session

Class to start an OAuth2 session.

Source code in teledetection/sdk/oauth2.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
class OAuth2Session:
    """Class to start an OAuth2 session."""

    def __init__(self, grant_type: type[GrantMethodBase] = DeviceGrant):
        """Initialize."""
        self.grant = grant_type()
        self.jwt_ttl_margin_seconds = 60
        self.jwt_issuance = datetime.datetime(year=1, month=1, day=1)
        self.jwt: JWT | None = None

    def save_token(self, now: datetime.datetime):
        """Save the JWT to disk."""
        if self.jwt:
            self.jwt_issuance = now
            self.jwt.to_config_dir()
        else:
            log.warning("No OAuth2 credentials to save")

    def _init_jwt(self):
        """Initialize the JWT."""
        if not self.jwt:
            # First JWT initialisation
            self.jwt = JWT.from_config_dir()
        if not self.jwt:
            # When JWT is still `None`, we use the grant method
            self.jwt = self.grant.get_first_token()
            self.save_token(datetime.datetime.now())

    def _refresh_if_needed(self):
        """Refresh the token if ttl is too short."""
        self._init_jwt()
        ttl_margin_seconds = 30
        now = datetime.datetime.now()
        jwt_expires_in = datetime.timedelta(seconds=self.jwt.expires_in)
        access_token_ttl = self.jwt_issuance + jwt_expires_in - now
        access_token_ttl_seconds = access_token_ttl.total_seconds()
        log.debug("access_token_ttl is %s", access_token_ttl_seconds)
        if access_token_ttl_seconds < ttl_margin_seconds:
            # Access token in not valid, but refresh might be
            try:
                self.jwt = self.grant.refresh_token(self.jwt)
            except RefreshTokenError as con_err:
                log.warning(  # pragma: no cover
                    "Unable to refresh token (reason: %s). "
                    "Renewing initial authentication.",
                    con_err,
                )
                self.jwt = self.grant.get_first_token()  # pragma: no cover
        else:
            # Token is still valid
            log.debug("Credentials still valid")
            return
        self.save_token(now)

    def get_access_token(self) -> str:
        """Return the access token."""
        self._init_jwt()
        self._refresh_if_needed()
        assert self.jwt
        return self.jwt.access_token

__init__(grant_type=DeviceGrant)

Initialize.

Source code in teledetection/sdk/oauth2.py 
157
158
159
160
161
162
def __init__(self, grant_type: type[GrantMethodBase] = DeviceGrant):
    """Initialize."""
    self.grant = grant_type()
    self.jwt_ttl_margin_seconds = 60
    self.jwt_issuance = datetime.datetime(year=1, month=1, day=1)
    self.jwt: JWT | None = None

get_access_token()

Return the access token.

Source code in teledetection/sdk/oauth2.py 
208
209
210
211
212
213
def get_access_token(self) -> str:
    """Return the access token."""
    self._init_jwt()
    self._refresh_if_needed()
    assert self.jwt
    return self.jwt.access_token

save_token(now)

Save the JWT to disk.

Source code in teledetection/sdk/oauth2.py 
164
165
166
167
168
169
170
def save_token(self, now: datetime.datetime):
    """Save the JWT to disk."""
    if self.jwt:
        self.jwt_issuance = now
        self.jwt.to_config_dir()
    else:
        log.warning("No OAuth2 credentials to save")

RefreshTokenError

Bases: Exception

Unable to refresh token.

Source code in teledetection/sdk/oauth2.py 
33
34
class RefreshTokenError(Exception):
    """Unable to refresh token."""

retrieve_token_endpoint()

Retrieve the token endpoint from the s3 signing endpoint.

Source code in teledetection/sdk/oauth2.py 
20
21
22
23
24
25
26
27
28
29
30
def retrieve_token_endpoint():
    """Retrieve the token endpoint from the s3 signing endpoint."""
    openapi_url = urljoin(ENV.tld_signing_endpoint, "openapi.json")
    log.debug("Fetching OAuth2 endpoint from openapi url %s", openapi_url)
    _session = create_session()
    res = _session.get(openapi_url, timeout=TIMEOUT)
    res.raise_for_status()
    data = res.json()
    return data["components"]["securitySchemes"]["OAuth2PasswordBearer"]["flows"][
        "password"
    ]["tokenUrl"]