skills/agentrhq__authsome
1
0
Fork 0
mirror of https://github.com/agentrhq/authsome.git synced 2026-09-19 01:34:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: metadata doesnt need separate file

Browse Source
This commit is contained in:
Manoj Bajaj
2026-06-04 19:34:15 +05:30
parent cf95c6c5da
commit 3ce646469a
6 changed files with 56 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/authsome/cli/client.py
+48 -44
View File
@@ -15,12 +15,8 @@ from urllib.parse import urlparse
import httpx
from authsome.cli.identity import (
IdentitySource,
RuntimeIdentity,
load_identity,
load_private_key,
load_runtime_identity,
mark_registered,
)
from authsome.config import get_authsome_config
from authsome.identity.proof import POP_AUTH_SCHEME, create_proof_jwt
@@ -59,6 +55,10 @@ def raise_for_error(response: httpx.Response) -> None:
obj = None
try:
data = response.json()
if response.status_code == 401 and data.get("detail") == "Unknown identity handle":
import authsome.errors as err_mod
raise err_mod.IdentityNotRegisteredError("current identity") from exc
error_name = data.get("error")
message = data.get("message")
if error_name and message:
@@ -70,6 +70,8 @@ def raise_for_error(response: httpx.Response) -> None:
Exception.__init__(obj, message)
obj.provider = data.get("provider")
obj.operation = data.get("operation")
except httpx.HTTPStatusError:
raise
except Exception:
pass
@@ -91,6 +93,7 @@ class AuthsomeApiClient:
self._base_url = (base_url or resolve_daemon_url()).rstrip("/")
self._home = home or get_authsome_config().home
self._identity = identity
self._server_registered = False # in-memory flag; reset on 401
@property
def base_url(self) -> str:
@@ -104,6 +107,7 @@ class AuthsomeApiClient:
body: dict[str, Any] | None = None,
timeout: int = 30,
protected: bool = True,
_retry: bool = True,
) -> dict[str, Any]:
body_bytes = b""
headers: dict[str, str] = {}
@@ -119,6 +123,15 @@ class AuthsomeApiClient:
content=body_bytes if body is not None else None,
headers=headers,
)
if protected and _retry and response.status_code == 401:
try:
detail = response.json().get("detail", "")
except Exception:
detail = ""
if detail == "Unknown identity handle":
self._server_registered = False
await self.ensure_identity_ready()
return await self._request(method, path, body=body, timeout=timeout, protected=protected, _retry=False)
raise_for_error(response)
return response.json()
@@ -127,15 +140,6 @@ class AuthsomeApiClient:
self._identity = load_runtime_identity(self._home)
return self._identity
def _filesystem_runtime_for_handle(self, handle: str) -> RuntimeIdentity:
identity = load_identity(self._home, handle)
return RuntimeIdentity(
handle=identity.handle,
did=identity.did,
source=IdentitySource.FILESYSTEM,
signer=load_private_key(self._home, identity.handle),
)
async def _proof_headers(self, method: str, path: str, body: bytes) -> dict[str, str]:
identity = await self.ensure_identity_ready()
token = create_proof_jwt(
@@ -149,52 +153,52 @@ class AuthsomeApiClient:
return {"Authorization": f"{POP_AUTH_SCHEME} {token}"}
async def ensure_identity_ready(self) -> RuntimeIdentity:
"""Ensure the acting identity is registered and claimed by a principal.
"""Ensure the acting identity is registered with the server and claimed.
A freshly registered identity must be claimed by a principal before it
can make authenticated calls; the daemon returns a browser claim URL
which is opened here while we poll for completion.
Checks server status on the first call per client instance (cached in
memory after that). Registers and opens the browser claim URL when the
identity is new or the server has been reset.
"""
runtime = self._runtime_identity()
if runtime.source is IdentitySource.ENV:
return await self._ensure_env_identity_ready(runtime)
identity = load_identity(self._home, runtime.handle)
if not identity.registered_for(self._base_url):
await self.register_identity(identity.handle, identity.did)
identity = mark_registered(self._home, identity.handle, server_url=self._base_url)
else:
if self._server_registered:
return runtime
await self._check_server_registration(runtime)
self._server_registered = True
return runtime
self._identity = self._filesystem_runtime_for_handle(identity.handle)
return self._identity
async def _ensure_env_identity_ready(self, identity: RuntimeIdentity) -> RuntimeIdentity:
async def _check_server_registration(self, runtime: RuntimeIdentity) -> None:
"""Verify registration with the server; register and claim if needed."""
try:
status = await self.get_identity_status(identity.handle)
except Exception:
status = await self.register_identity(identity.handle, identity.did)
status = await self.get_identity_status(runtime.handle)
except httpx.HTTPStatusError as exc:
if exc.response.status_code != 404:
raise
status = await self.register_identity(runtime.handle, runtime.did)
registration_status = status.get("registration_status", "registered")
if registration_status == "unknown":
await self.register_identity(identity.handle, identity.did)
return identity
reg_status = status.get("registration_status", "")
if reg_status == "claim_required":
claim_url = status.get("claim_url", "")
if claim_url:
self._open_claim_url(claim_url)
await self._poll_claim_completion(runtime.handle)
elif reg_status == "rejected":
raise RuntimeError(f"Identity '{runtime.handle}' claim was rejected by the server")
def _open_claim_url(self, claim_url: str) -> None:
"""Surface the browser claim URL (so headless users can open it) and try to launch it."""
print(
f"Open this URL in your browser to register and claim this identity:\n {claim_url}",
file=sys.stderr,
)
print(f"Open this URL in your browser to claim this identity:\n {claim_url}", file=sys.stderr)
with suppress(Exception):
webbrowser.open(claim_url)
async def _poll_claim_completion(self, handle: str, *, timeout_seconds: int = 300) -> dict[str, Any]:
async def _poll_claim_completion(self, handle: str, *, timeout_seconds: int = 300) -> None:
print("Waiting for identity to be claimed...", file=sys.stderr)
deadline = asyncio.get_running_loop().time() + timeout_seconds
while True:
status = await self.get_identity_status(handle)
if status.get("registration_status") in {"claimed", "registered"}:
return status
reg_status = status.get("registration_status", "")
if reg_status == "claimed":
return
if reg_status == "rejected":
raise RuntimeError(f"Identity '{handle}' claim was rejected")
if asyncio.get_running_loop().time() >= deadline:
raise TimeoutError(f"Timed out waiting for identity '{handle}' to be claimed")
await asyncio.sleep(1)
src/authsome/cli/commands/profile.py
+1 -1
View File
@@ -27,7 +27,7 @@ async def profile_create(ctx_obj: ContextObj, handle: str | None) -> None:
"home": str(home),
"profile": identity_meta.handle,
"did": identity_meta.did,
"registration_status": "registered" if identity_meta.registered else "local",
"registration_status": "local",
"switched": True,
}
ctx_obj.print_json(data)
src/authsome/cli/identity.py
-40
View File
@@ -5,7 +5,6 @@ from __future__ import annotations
import os
from collections.abc import Mapping
from contextlib import suppress
from datetime import UTC, datetime
from enum import StrEnum
from pathlib import Path
@@ -15,10 +14,8 @@ from pydantic import BaseModel
from authsome.cli.config import load_client_config, save_client_config
from authsome.identity.helpers import (
IdentityMetadata,
IdentityStatus,
create_identity_material,
generate_handle,
normalize_server_url,
private_key_from_hex,
private_key_to_hex,
public_key_to_did_key,
@@ -98,43 +95,6 @@ def remove_legacy_default_identity(home: Path) -> None:
path.unlink()
def mark_registered(home: Path, handle: str, *, server_url: str | None = None) -> IdentityMetadata:
"""Persist a registered state for a local identity after daemon registration."""
metadata = load_identity(home, handle)
registered_servers = list(metadata.registered_servers)
if server_url is not None:
normalized_server_url = normalize_server_url(server_url)
if normalized_server_url not in registered_servers:
registered_servers.append(normalized_server_url)
if metadata.identity_status == IdentityStatus.CLAIMED:
if registered_servers == metadata.registered_servers:
return metadata
updated = metadata.model_copy(
update={"registered_servers": registered_servers, "updated_at": datetime.now(UTC)}
)
identity_metadata_path(home, handle).write_text(updated.model_dump_json(indent=2), encoding="utf-8")
return updated
if server_url is None:
registered_servers = metadata.registered_servers
updated = metadata.model_copy(
update={
"identity_status": IdentityStatus.REGISTERED,
"registered_servers": registered_servers,
"updated_at": datetime.now(UTC),
}
)
identity_metadata_path(home, handle).write_text(updated.model_dump_json(indent=2), encoding="utf-8")
return updated
def mark_claimed(home: Path, handle: str) -> IdentityMetadata:
"""Persist a claimed state for a local identity after ownership resolution."""
metadata = load_identity(home, handle)
updated = metadata.model_copy(update={"identity_status": IdentityStatus.CLAIMED, "updated_at": datetime.now(UTC)})
identity_metadata_path(home, handle).write_text(updated.model_dump_json(indent=2), encoding="utf-8")
return updated
def ensure_local_identity(home: Path, active_handle: str | None = None) -> IdentityMetadata:
"""Return the active local identity, creating one if none exists."""
remove_legacy_default_identity(home)
src/authsome/errors.py
+7
View File
@@ -161,3 +161,10 @@ class EndpointUnreachableError(AuthsomeError):
def __init__(self, endpoint: str, reason: str = "Unreachable", *, provider: str | None = None) -> None:
super().__init__(f"Endpoint '{endpoint}' is unreachable: {reason}", provider=provider)
class IdentityNotRegisteredError(AuthsomeError):
"""Raised when the identity is not registered with the target server."""
def __init__(self, handle: str) -> None:
super().__init__(f"Identity '{handle}' is not registered with the server", operation="registration")
src/authsome/identity/__init__.py
-4
View File
@@ -3,10 +3,8 @@
from authsome.identity.helpers import (
IdentityMaterial,
IdentityMetadata,
IdentityStatus,
create_identity_material,
generate_handle,
normalize_server_url,
private_key_from_hex,
private_key_to_hex,
public_key_from_did_key,
@@ -33,7 +31,6 @@ __all__ = [
"IdentityClaimRecord",
"IdentityMaterial",
"IdentityMetadata",
"IdentityStatus",
"IdentityRegistration",
"PrincipalRecord",
"POP_AUTH_SCHEME",
@@ -43,7 +40,6 @@ __all__ = [
"create_identity_material",
"create_proof_jwt",
"generate_handle",
"normalize_server_url",
"private_key_from_hex",
"private_key_to_hex",
"public_key_from_did_key",
src/authsome/identity/helpers.py
-27
View File
@@ -6,7 +6,6 @@ import random
import re
from dataclasses import dataclass
from datetime import UTC, datetime
from enum import StrEnum
import base58
from cryptography.hazmat.primitives import serialization
@@ -50,35 +49,14 @@ _ADVERBS = (
)
class IdentityStatus(StrEnum):
"""Progressive local lifecycle state for an identity."""
UNREGISTERED = "unregistered"
REGISTERED = "registered"
CLAIMED = "claimed"
class IdentityMetadata(BaseModel):
"""Identity metadata associated with a caller-owned private key."""
handle: str
did: str
identity_status: IdentityStatus = IdentityStatus.UNREGISTERED
registered_servers: list[str] = Field(default_factory=list)
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
@property
def registered(self) -> bool:
return self.identity_status in {IdentityStatus.REGISTERED, IdentityStatus.CLAIMED}
@property
def claimed(self) -> bool:
return self.identity_status == IdentityStatus.CLAIMED
def registered_for(self, server_url: str) -> bool:
return normalize_server_url(server_url) in self.registered_servers
@dataclass(frozen=True)
class IdentityMaterial:
@@ -106,11 +84,6 @@ def validate_handle(handle: str) -> str:
return handle
def normalize_server_url(server_url: str) -> str:
"""Normalize daemon URLs before using them as local identity status keys."""
return server_url.rstrip("/") or server_url
def public_key_to_did_key(public_key: Ed25519PublicKey) -> str:
raw = public_key.public_bytes(
encoding=serialization.Encoding.Raw,