ligo.at
decentralized and customizable links page on top of atproto
ligo.at
atproto
link-in-bio
python
uv
resolve handles via well-known
nauta.one
+35
-10
src/atproto/__init__.py
+35
-10
src/atproto/__init__.py
···
1
1
from aiodns import DNSResolver, error as dns_error
2
2
from aiohttp.client import ClientSession
3
+
from asyncio import tasks
3
4
from os import getenv
4
5
from re import match as regex_match
5
6
from typing import Any
···
35
36
36
37
if is_valid_handle(query):
37
38
handle = query.lower()
38
-
did = await resolve_did_from_handle(handle, didkv)
39
+
did = await resolve_did_from_handle(client, handle, didkv)
39
40
if not did:
40
41
return None
41
42
doc = await resolve_doc_from_did(client, did)
···
54
55
handle = handle_from_doc(doc)
55
56
if not handle:
56
57
return None
57
-
if await resolve_did_from_handle(handle, didkv) != did:
58
+
if await resolve_did_from_handle(client, handle, didkv) != did:
58
59
return None
59
60
return (did, handle, doc)
60
61
···
73
74
74
75
75
76
async def resolve_did_from_handle(
77
+
client: ClientSession,
76
78
handle: str,
77
79
kv: KV = nokv,
78
80
reload: bool = False,
79
81
) -> str | None:
80
-
"""Returns the DID for a given handle"""
82
+
"""Returns the DID for a given handle."""
81
83
82
84
if not is_valid_handle(handle):
83
85
return None
···
86
88
if did is not None and not reload:
87
89
return did
88
90
89
-
resolver = DNSResolver()
90
-
try:
91
-
result = await resolver.query(f"_atproto.{handle}", "TXT")
92
-
except dns_error.DNSError:
93
-
return None
91
+
did = await _resolve_did_from_handle_dns(handle)
92
+
if did is None:
93
+
did = await _resolve_did_from_handle_wk(client, handle)
94
+
95
+
if did is not None and is_valid_did(did):
96
+
kv.set(handle, value=did)
97
+
return did
98
+
99
+
return None
100
+
101
+
102
+
async def _resolve_did_from_handle_wk(client: ClientSession, handle: str) -> str | None:
103
+
"""Resolve the DID for a given handle via .well-known"""
104
+
105
+
url = f"https://{handle}/.well-known/atproto-did"
106
+
response = await client.get(url)
107
+
if response.ok:
108
+
return await response.text()
109
+
return None
110
+
111
+
112
+
async def _resolve_did_from_handle_dns(handle: str) -> str | None:
113
+
"""Resolve the DID for a given handle via DNS."""
114
+
115
+
async with DNSResolver() as resolver:
116
+
try:
117
+
result = await resolver.query(f"_atproto.{handle}", "TXT")
118
+
except dns_error.DNSError:
119
+
return None
94
120
95
121
for record in result:
96
-
value = str(record.text).replace('"', "")
122
+
value = str(record.text).strip('"')
97
123
if value.startswith("did="):
98
124
did = value[4:]
99
125
if is_valid_did(did):
100
-
kv.set(handle, value=did)
101
126
return did
102
127
103
128
return None