ligo.at
decentralized and customizable links page on top of atproto
ligo.at
atproto
link-in-bio
python
uv
nauta.one
1from datetime import datetime, timedelta, timezone
2from typing import NamedTuple, TypeVar
3
4from aiohttp.client import ClientSession
5from authlib.jose import JsonWebKey
6from flask import current_app, request
7from flask.sessions import SessionMixin
8
9from src.atproto.oauth import refresh_token_request
10from src.atproto.types import OAuthAuthRequest, OAuthSession
11
12
13def save_auth_request(session: SessionMixin, request: OAuthAuthRequest):
14 return _set_into_session(session, "oauth_auth_request", request)
15
16
17def save_auth_session(session: SessionMixin, auth_session: OAuthSession):
18 return _set_into_session(session, "oauth_auth_session", auth_session)
19
20
21def delete_auth_request(session: SessionMixin):
22 return _delete_from_session(session, "oauth_auth_request")
23
24
25def delete_auth_session(session: SessionMixin):
26 return _delete_from_session(session, "oauth_auth_session")
27
28
29def get_auth_request(session: SessionMixin) -> OAuthAuthRequest | None:
30 return _get_from_session(session, "oauth_auth_request", OAuthAuthRequest)
31
32
33def get_auth_session(session: SessionMixin) -> OAuthSession | None:
34 return _get_from_session(session, "oauth_auth_session", OAuthSession)
35
36
37def _set_into_session(session: SessionMixin, key: str, value: NamedTuple):
38 session[key] = value._asdict()
39
40
41def _delete_from_session(session: SessionMixin, key: str):
42 try:
43 del session[key]
44 except KeyError:
45 pass
46
47
48async def refresh_auth_session(
49 session: SessionMixin,
50 client: ClientSession,
51 current: OAuthSession,
52) -> OAuthSession | None:
53 current_app.logger.debug("refreshing oauth tokens")
54 CLIENT_SECRET_JWK = JsonWebKey.import_key(current_app.config["CLIENT_SECRET_JWK"])
55 tokens, dpop_authserver_nonce = await refresh_token_request(
56 client=client,
57 user=current,
58 app_host=request.host,
59 client_secret_jwk=CLIENT_SECRET_JWK,
60 )
61 now = datetime.now(timezone.utc)
62 expires_at = now + timedelta(seconds=tokens.expires_in or 300)
63 user = current._replace(
64 access_token=tokens.access_token,
65 refresh_token=tokens.refresh_token,
66 expires_at=int(expires_at.timestamp()),
67 dpop_pds_nonce=dpop_authserver_nonce,
68 )
69 save_auth_session(session, user)
70 return user
71
72
73OAuthClass = TypeVar("OAuthClass")
74
75
76def _get_from_session(
77 session: SessionMixin,
78 key: str,
79 Type: type[OAuthClass],
80) -> OAuthClass | None:
81 if key not in session:
82 return None
83
84 try:
85 return Type(**session[key])
86 except TypeError as exception:
87 current_app.logger.debug(f"unable to load {key}")
88 current_app.logger.debug(exception)
89 del session[key]
90 return None