auth: Add get_id_token()

This will optionally refresh the ID token and return it for use in the
Authorization HTTP header.

not_my_board/_auth/__init__.py

@@ -1 +1 @@
-from ._login import LoginFlow
+from ._login import LoginFlow, get_id_token

not_my_board/_auth/_login.py

@@ -6,7 +6,7 @@
 import not_my_board._jsonrpc as jsonrpc
 import not_my_board._util as util
 
-from ._openid import AuthRequest
+from ._openid import AuthRequest, ensure_fresh
 
 
 class LoginFlow(util.ContextStack):
@@ -71,6 +71,17 @@ async def oidc_callback_registered(self):
         self._ready_event.set()
 
 
+async def get_id_token(hub_url, http_client):
+    async with _TokenStore() as token_store:
+        id_token, refresh_token = token_store.get_tokens(hub_url)
+        id_token, refresh_token = await ensure_fresh(
+            id_token, refresh_token, http_client
+        )
+        token_store.save_tokens(hub_url, id_token, refresh_token)
+
+    return id_token
+
+
 class _TokenStore(util.ContextStack):
     _path = pathlib.Path("/var/lib/not-my-board/auth_tokens.json")
 

not_my_board/_auth/_openid.py

@@ -109,6 +109,41 @@ async def request_tokens(self, auth_response, http_client):
         return response["id_token"], response["refresh_token"], claims
 
 
+async def ensure_fresh(id_token, refresh_token, http_client):
+    if _needs_refresh(id_token):
+        claims = jwt.decode(id_token, options={"verify_signature": False})
+        issuer_url = claims["iss"]
+        client_id = claims["aud"]
+        identity_provider = await IdentityProvider.from_url(issuer_url, http_client)
+
+        params = {
+            "grant_type": "refresh_token",
+            "refresh_token": refresh_token,
+            "client_id": client_id,
+        }
+        response = await http_client.post_form(identity_provider.token_endpoint, params)
+        return response["id_token"], response["refresh_token"]
+    else:
+        return id_token, refresh_token
+
+
+def _needs_refresh(id_token):
+    try:
+        jwt.decode(
+            id_token,
+            options={
+                "verify_signature": False,
+                "require": ["iss", "sub", "aud", "exp", "iat"],
+                "verify_exp": True,
+                "verify_iat": True,
+                "verify_nbf": True,
+            },
+        )
+    except Exception:
+        return True
+    return False
+
+
 async def verify(token, client_id, http_client):
     unverified_token = jwt.api_jwt.decode_complete(
         token, options={"verify_signature": False}