···2from typing import Any
3import json
45-from .atproto import PdsUrl, get_record, resolve_did_from_handle, resolve_pds_from_did
0000006from .atproto.oauth import pds_authed_req
7from .db import close_db_connection, init_db
8from .oauth import get_auth_session, oauth, save_auth_session
···16links: dict[str, list[dict[str, str]]] = {}
17profiles: dict[str, tuple[str, str]] = {}
1819-SCHEMA = "one.nauta"
202122@app.before_request
···38 return render_template("index.html")
39400000000041@app.get("/@<string:handle>")
42-def page_profile(handle: str):
43 reload = request.args.get("reload") is not None
4445 did = resolve_did_from_handle(handle, reload=reload)
46 if did is None:
47 return "did not found", 404
000048 pds = resolve_pds_from_did(did, reload=reload)
49 if pds is None:
50 return "pds not found", 404
···242 user=user,
243 update_dpop_pds_nonce=update_dpop_pds_nonce,
244 )
245- if not response or not response.ok:
246 app.logger.warning("PDS HTTP ERROR")
···2from typing import Any
3import json
45+from .atproto import (
6+ PdsUrl,
7+ get_record,
8+ is_valid_did,
9+ resolve_did_from_handle,
10+ resolve_pds_from_did,
11+)
12from .atproto.oauth import pds_authed_req
13from .db import close_db_connection, init_db
14from .oauth import get_auth_session, oauth, save_auth_session
···22links: dict[str, list[dict[str, str]]] = {}
23profiles: dict[str, tuple[str, str]] = {}
2425+SCHEMA = "at.ligo"
262728@app.before_request
···44 return render_template("index.html")
454647+@app.get("/did:<string:did>")
48+def page_profile_with_did(did: str):
49+ did = f"did:{did}"
50+ if not is_valid_did(did):
51+ return "invalid did", 400
52+ return page_profile(did)
53+54+55@app.get("/@<string:handle>")
56+def page_profile_with_handle(handle: str):
57 reload = request.args.get("reload") is not None
5859 did = resolve_did_from_handle(handle, reload=reload)
60 if did is None:
61 return "did not found", 404
62+ return page_profile(did, reload=reload)
63+64+65+def page_profile(did: str, reload: bool = False):
66 pds = resolve_pds_from_did(did, reload=reload)
67 if pds is None:
68 return "pds not found", 404
···260 user=user,
261 update_dpop_pds_nonce=update_dpop_pds_nonce,
262 )
263+ if not response or not response.is_success:
264 app.logger.warning("PDS HTTP ERROR")
+13-11
src/security.py
···1from urllib.parse import urlparse
2-import requests_hardened
345# this is a crude/partial filter that looks at HTTPS URLs and checks if they seem "safe" for server-side requests (SSRF). This is only a partial mitigation, the actual HTTP client also needs to prevent other attacks and behaviors.
···29 return True
303132-# configures a "hardened" requests wrapper
33-hardened_http = requests_hardened.Manager(
34- requests_hardened.Config(
35- default_timeout=(2, 10),
36- never_redirect=True,
37- ip_filter_enable=True,
38- ip_filter_allow_loopback_ips=False,
39- user_agent_override="AtprotoCookbookOAuthFlaskDemo",
40- )
41-)
00
···1from urllib.parse import urlparse
2+import httpx
345# this is a crude/partial filter that looks at HTTPS URLs and checks if they seem "safe" for server-side requests (SSRF). This is only a partial mitigation, the actual HTTP client also needs to prevent other attacks and behaviors.
···29 return True
303132+class HardenedHttp:
33+ def get_session(self) -> httpx.Client:
34+ return httpx.Client(
35+ timeout=httpx.Timeout(20, connect=5),
36+ follow_redirects=False,
37+ headers={
38+ "User-Agent": "ligo.at/0",
39+ },
40+ )
41+42+43+hardened_http = HardenedHttp()