ligo.at
nauta.one
+13
-11
src/atproto/__init__.py
+13
-11
src/atproto/__init__.py
···
1
-
from aiodns import DNSResolver, error as dns_error
2
-
from aiohttp.client import ClientSession
3
-
from asyncio import tasks
4
from os import getenv
5
from re import match as regex_match
6
from typing import Any
7
8
from .kv import KV, nokv
9
from .validator import is_valid_authserver_meta
10
-
from ..security import is_safe_url
11
12
PLC_DIRECTORY = getenv("PLC_DIRECTORY_URL") or "https://plc.directory"
13
HANDLE_REGEX = r"^([a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?$"
···
163
) -> dict[str, Any] | None:
164
"""Returns the DID document"""
165
166
if did.startswith("did:plc:"):
167
response = await client.get(f"{PLC_DIRECTORY}/{did}")
168
-
if response.ok:
169
-
return await response.json()
170
-
return None
171
172
-
if did.startswith("did:web:"):
173
-
# TODO: resolve did:web
174
-
raise Exception("resolve did:web")
175
-
176
return None
177
178
···
1
from os import getenv
2
from re import match as regex_match
3
from typing import Any
4
5
+
from aiodns import DNSResolver
6
+
from aiodns import error as dns_error
7
+
from aiohttp.client import ClientResponse, ClientSession
8
+
9
+
from ..security import is_safe_url
10
from .kv import KV, nokv
11
from .validator import is_valid_authserver_meta
12
13
PLC_DIRECTORY = getenv("PLC_DIRECTORY_URL") or "https://plc.directory"
14
HANDLE_REGEX = r"^([a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?$"
···
164
) -> dict[str, Any] | None:
165
"""Returns the DID document"""
166
167
+
response: ClientResponse | None = None
168
+
169
if did.startswith("did:plc:"):
170
response = await client.get(f"{PLC_DIRECTORY}/{did}")
171
+
elif did.startswith("did:web:"):
172
+
domain = did.split(":")[-1]
173
+
assert is_valid_handle(domain)
174
+
response = await client.get(f"https://{domain}/.well-known/did.json")
175
176
+
if response and response.ok:
177
+
return await response.json()
178
return None
179
180