ligo.at
1import asyncio
2import json
3from typing import Any, NamedTuple
4
5from aiohttp.client import ClientSession
6from flask import Flask, g, redirect, render_template, request, session, url_for
7from flask_htmx import HTMX
8from flask_htmx import make_response as htmx_response
9
10from src.atproto import (
11 PdsUrl,
12 get_record,
13 is_valid_did,
14 resolve_did_from_handle,
15 resolve_pds_from_did,
16)
17from src.atproto.oauth import pds_authed_req
18from src.atproto.types import OAuthSession
19from src.auth import get_auth_session, save_auth_session
20from src.db import KV, close_db_connection, get_db, init_db
21from src.oauth import oauth
22
23app = Flask(__name__)
24_ = app.config.from_prefixed_env()
25app.register_blueprint(oauth)
26htmx = HTMX()
27htmx.init_app(app)
28init_db(app)
29
30
31@app.before_request
32async def load_user_to_context():
33 g.user = get_auth_session(session)
34
35
36def get_user() -> OAuthSession | None:
37 return g.user
38
39
40@app.teardown_appcontext
41async def app_teardown(exception: BaseException | None):
42 close_db_connection(exception)
43
44
45@app.get("/")
46def page_home():
47 return render_template("index.html")
48
49
50@app.get("/<string:atid>")
51async def page_profile(atid: str):
52 reload = request.args.get("reload") is not None
53
54 db = get_db(app)
55 didkv = KV(db, app.logger, "did_from_handle")
56 pdskv = KV(db, app.logger, "pds_from_did")
57
58 async with ClientSession() as client:
59 if atid.startswith("@"):
60 handle = atid[1:].lower()
61 did = await resolve_did_from_handle(client, handle, kv=didkv, reload=reload)
62 if did is None:
63 return render_template("error.html", message="did not found"), 404
64 elif is_valid_did(atid):
65 handle = None
66 did = atid
67 else:
68 return render_template("error.html", message="invalid did or handle"), 400
69
70 if _is_did_blocked(did):
71 return render_template("error.html", message="profile not found"), 404
72
73 pds = await resolve_pds_from_did(client, did=did, kv=pdskv, reload=reload)
74 if pds is None:
75 return render_template("error.html", message="pds not found"), 404
76 (profile, _), link_sections = await asyncio.gather(
77 load_profile(client, pds, did, reload=reload),
78 load_links(client, pds, did, reload=reload),
79 )
80 if profile is None or link_sections is None:
81 return render_template("error.html", message="profile not found"), 404
82
83 if reload:
84 # remove the ?reload parameter
85 return redirect(request.path)
86
87 if handle:
88 profile["handle"] = handle
89 athref = f"at://{did}/at.ligo.actor.links/self"
90 return render_template(
91 "profile.html",
92 profile=profile,
93 links=link_sections[0].links,
94 sections=link_sections,
95 athref=athref,
96 )
97
98
99@app.get("/login")
100def page_login():
101 if get_user() is not None:
102 return redirect("/editor")
103 return render_template("login.html")
104
105
106@app.post("/login")
107def auth_login():
108 value = request.form.get("username") or request.form.get("authserver")
109 if value and value[0] == "@":
110 value = value[1:]
111 if not value:
112 return redirect(url_for("page_login"), 303)
113 return redirect(url_for("oauth.oauth_start", username_or_authserver=value), 303)
114
115
116@app.route("/auth/logout")
117def auth_logout():
118 session.clear()
119 return redirect(url_for("page_login"), 303)
120
121
122@app.get("/editor")
123async def page_editor():
124 user = get_user()
125 if user is None:
126 return redirect("/login", 302)
127
128 did: str = user.did
129 pds: str = user.pds_url
130 handle: str | None = user.handle
131
132 async with ClientSession() as client:
133 (profile, from_bluesky), link_sections = await asyncio.gather(
134 load_profile(client, pds, did),
135 load_links(client, pds, did),
136 )
137
138 links = []
139 if link_sections:
140 links = link_sections[0].links
141
142 return render_template(
143 "editor.html",
144 handle=handle,
145 profile=profile,
146 profile_from_bluesky=from_bluesky,
147 links=json.dumps(links),
148 )
149
150
151@app.post("/editor/profile")
152async def post_editor_profile():
153 user = get_user()
154 if user is None:
155 url = url_for("auth_logout")
156 return htmx_response(redirect=url) if htmx else redirect(url, 303)
157
158 display_name = request.form.get("displayName")
159 description = request.form.get("description", "")
160 if not display_name:
161 return redirect("/editor", 303)
162
163 record = {
164 "$type": "at.ligo.actor.profile",
165 "displayName": display_name,
166 "description": description,
167 }
168
169 success = await put_record(
170 user=user,
171 pds=user.pds_url,
172 repo=user.did,
173 collection="at.ligo.actor.profile",
174 rkey="self",
175 record=record,
176 )
177
178 if success:
179 kv = KV(app, app.logger, "profile_from_did")
180 kv.set(user.did, json.dumps(record))
181 else:
182 app.logger.warning("log out user for now")
183 url = url_for("auth_logout")
184 return htmx_response(redirect=url) if htmx else redirect(url, 303)
185
186 if htmx:
187 return htmx_response(
188 render_template("_editor_profile.html", profile=record),
189 reswap="outerHTML",
190 )
191
192 return redirect(url_for("page_editor"), 303)
193
194
195@app.post("/editor/links")
196async def post_editor_links():
197 user = get_user()
198 if user is None:
199 url = url_for("auth_logout")
200 return htmx_response(redirect=url) if htmx else redirect(url, 303)
201
202 links: list[dict[str, str]] = []
203 hrefs = request.form.getlist("link-href")
204 titles = request.form.getlist("link-title")
205 subtitles = request.form.getlist("link-subtitle")
206 backgrounds = request.form.getlist("link-background-color")
207 for href, title, subtitle, background in zip(hrefs, titles, subtitles, backgrounds):
208 if not href or not title or not background:
209 break
210 link: dict[str, str] = {
211 "href": href,
212 "title": title,
213 "backgroundColor": background,
214 }
215 if subtitle:
216 link["subtitle"] = subtitle
217 links.append(link)
218
219 record = {
220 "$type": "at.ligo.actor.links",
221 "sections": [
222 {
223 "title": "",
224 "links": links,
225 }
226 ],
227 }
228
229 success = await put_record(
230 user=user,
231 pds=user.pds_url,
232 repo=user.did,
233 collection="at.ligo.actor.links",
234 rkey="self",
235 record=record,
236 )
237
238 if success:
239 kv = KV(app, app.logger, "links_from_did")
240 kv.set(user.did, json.dumps(record))
241 else:
242 app.logger.warning("log out user for now")
243 url = url_for("auth_logout")
244 return htmx_response(redirect=url) if htmx else redirect(url, 303)
245
246 if htmx:
247 return htmx_response(
248 render_template("_editor_links.html", links=record["sections"][0]["links"]),
249 reswap="outerHTML",
250 )
251
252 return redirect(url_for("page_editor"), 303)
253
254
255@app.get("/terms")
256def page_terms():
257 return render_template("terms.html")
258
259
260class LinkSection(NamedTuple):
261 title: str
262 links: list[dict[str, str]]
263
264
265async def load_links(
266 client: ClientSession,
267 pds: str,
268 did: str,
269 reload: bool = False,
270) -> list[LinkSection] | None:
271 kv = KV(app, app.logger, "links_from_did")
272 record_json = kv.get(did)
273
274 if record_json is not None and not reload:
275 parsed = json.loads(record_json)
276 return _links_or_sections(parsed)
277
278 record = await get_record(client, pds, did, "at.ligo.actor.links", "self")
279 if record is None:
280 return None
281
282 kv.set(did, value=json.dumps(record))
283 return _links_or_sections(record)
284
285
286def _links_or_sections(raw: dict[str, Any]) -> list[LinkSection] | None:
287 if "sections" in raw:
288 return list(map(lambda s: LinkSection(**s), raw["sections"]))
289 elif "links" in raw:
290 return [LinkSection("", raw["links"])]
291 else:
292 return None
293
294
295async def load_profile(
296 client: ClientSession,
297 pds: str,
298 did: str,
299 fallback_with_bluesky: bool = True,
300 reload: bool = False,
301) -> tuple[dict[str, str] | None, bool]:
302 kv = KV(app, app.logger, "profile_from_did")
303 record_json = kv.get(did)
304
305 if record_json is not None and not reload:
306 return json.loads(record_json), False
307
308 (record, bsky_record) = await asyncio.gather(
309 get_record(client, pds, did, "at.ligo.actor.profile", "self"),
310 get_record(client, pds, did, "app.bsky.actor.profile", "self"),
311 )
312
313 from_bluesky = False
314 if record is None and fallback_with_bluesky:
315 record = bsky_record
316 from_bluesky = True
317
318 if record is not None:
319 kv.set(did, value=json.dumps(record))
320
321 return record, from_bluesky
322
323
324# TODO: move to .atproto
325async def put_record(
326 user: OAuthSession,
327 pds: PdsUrl,
328 repo: str,
329 collection: str,
330 rkey: str,
331 record: dict[str, Any],
332) -> bool:
333 """Writes the record onto the users PDS. Returns bool for success."""
334
335 endpoint = f"{pds}/xrpc/com.atproto.repo.putRecord"
336 body = {
337 "repo": repo,
338 "collection": collection,
339 "rkey": rkey,
340 "record": record,
341 }
342
343 def update_dpop_pds_nonce(nonce: str):
344 session_ = user._replace(dpop_pds_nonce=nonce)
345 save_auth_session(session, session_)
346
347 response = await pds_authed_req(
348 method="POST",
349 url=endpoint,
350 body=body,
351 user=user,
352 update_dpop_pds_nonce=update_dpop_pds_nonce,
353 )
354
355 if not response.ok:
356 app.logger.warning(f"put_record failed with status {response.status}")
357 app.logger.warning(await response.text())
358
359 return response.ok
360
361
362def _is_did_blocked(did: str) -> bool:
363 kv = KV(app, app.logger, "blockeddids")
364 return kv.get(did) is not None