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: remove vault rekey functionality and associated endpoints

Browse Source
This commit is contained in:
Ankit Ranjan
2026-05-25 18:03:47 +05:30
parent 13a25f7c76
commit 7e3a63480c
8 changed files with 38 additions and 670 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/authsome/cli/admin.py
-1
View File
@@ -6,7 +6,6 @@ import sys
from pathlib import Path
import click
from loguru import logger
from authsome.cli.context import ContextObj
from authsome.cli.daemon_control import (
src/authsome/cli/client.py
-3
View File
@@ -283,9 +283,6 @@ class AuthsomeApiClient:
async def whoami(self) -> dict[str, Any]:
return await self._get("/whoami")
async def rekey(self) -> dict[str, Any]:
return await self._post("/rekey", {})
async def doctor(self) -> dict[str, Any]:
return await self.ready()
src/authsome/cli/main.py
+35 -305
View File
@@ -1,6 +1,6 @@
"""Command-line interface for authsome."""
import json as json_lib
import json
import os
import pathlib
import sys
@@ -25,7 +25,7 @@ from authsome.cli.helpers import (
setup_logging,
)
from authsome.paths import get_client_log_path
from authsome.utils import connection_is_active, format_error_code, format_expires_at, redact
from authsome.utils import connection_is_active, format_error_code, redact
@click.group()
@@ -50,15 +50,6 @@ def cli(ctx: click.Context, verbose: bool, log_file: str) -> None:
setup_logging(verbose=verbose, log_file=resolved)
def _render_encryption_backend(data: dict[str, Any]) -> str:
"""Render configured mode plus effective master-key source for human output."""
backend = data["encryption_backend"]
configured_mode = data.get("configured_encryption_mode")
if configured_mode:
return f"{backend} (mode: {configured_mode})"
return backend
@cli.group(name="provider")
def provider() -> None:
"""Manage provider definitions and provider-level operations."""
@@ -111,125 +102,7 @@ async def list_cmd(ctx_obj: ContextObj) -> None:
bundled_out = [build_provider_entry(p, "bundled") for p in by_source["bundled"]]
custom_out = [build_provider_entry(p, "custom") for p in by_source["custom"]]
if ctx_obj.json_output:
ctx_obj.print_json({"bundled": bundled_out, "custom": custom_out})
return
rows: list[dict[str, Any]] = []
for p in bundled_out + custom_out:
provider_label = f"{p['display_name']} [{p['name']}]"
if p["connections"]:
for conn in p["connections"]:
rows.append(
{
"provider_id": p["name"],
"provider": provider_label,
"source": p["source"],
"auth": p["auth_type"],
"connection": (
f"{conn['connection_name']} (default)"
if conn.get("is_default")
else conn["connection_name"]
),
"status": conn["status"],
"expires_at": conn.get("expires_at"),
"expires": format_expires_at(conn.get("expires_at")) or "-",
}
)
else:
rows.append(
{
"provider_id": p["name"],
"provider": provider_label,
"source": p["source"],
"auth": p["auth_type"],
"connection": "-",
"status": "not_connected",
"expires_at": None,
"expires": "-",
}
)
if not rows:
ctx_obj.echo("No providers configured.")
return
connected_provider_ids = {row["provider_id"] for row in rows if connection_is_active(row)}
connected_count = len(connected_provider_ids)
ctx_obj.echo(f"Providers: {len(bundled_out) + len(custom_out)} total, {connected_count} connected")
ctx_obj.echo("")
headers = {
"provider": "Provider",
"source": "Source",
"auth": "Auth",
"connection": "Connection",
"status": "Status",
"expires": "Expires",
}
widths = {
key: max(len(headers[key]), *(len(row[key]) for row in rows))
for key in ("provider", "source", "auth", "connection", "status", "expires")
}
def pad_field(text: str, key: str, color: str | None = None, bold: bool = False, dim: bool = False) -> str:
if ctx_obj.no_color or (not color and not dim):
return f"{text:<{widths[key]}}"
styled = click.style(text, fg=color, bold=bold, dim=dim)
padding = " " * (widths[key] - len(text))
return f"{styled}{padding}"
def render_row(row: dict[str, Any], is_header: bool = False, is_divider: bool = False) -> str:
if is_header or is_divider:
return (
f"{row['provider']:<{widths['provider']}} "
f"{row['source']:<{widths['source']}} "
f"{row['auth']:<{widths['auth']}} "
f"{row['connection']:<{widths['connection']}} "
f"{row['status']:<{widths['status']}} "
f"{row['expires']:<{widths['expires']}}"
).rstrip()
is_active = connection_is_active(row)
if is_active:
prov_color = "green"
prov_bold = True
conn_color = "cyan"
status_color = "green"
status_dim = False
expires_color = "yellow"
else:
prov_color = None
prov_bold = False
conn_color = None
expires_color = None
if row["status"] == "not_connected":
status_color = None
status_dim = True
else:
status_color = "red"
status_dim = False
provider_str = pad_field(row["provider"], "provider", color=prov_color, bold=prov_bold)
source_str = pad_field(row["source"], "source")
auth_str = pad_field(row["auth"], "auth")
connection_str = pad_field(row["connection"], "connection", color=conn_color)
status_str = pad_field(row["status"], "status", color=status_color, bold=is_active, dim=status_dim)
expires_str = pad_field(row["expires"], "expires", color=expires_color)
return f"{provider_str} {source_str} {auth_str} {connection_str} {status_str} {expires_str}".rstrip()
ctx_obj.emit(render_row(headers, is_header=True))
ctx_obj.emit(
render_row(
{key: "-" * widths[key] for key in ("provider", "source", "auth", "connection", "status", "expires")},
is_divider=True,
)
)
for row in rows:
ctx_obj.emit(render_row(row))
ctx_obj.print_json({"bundled": bundled_out, "custom": custom_out})
@cli.command()
@@ -259,11 +132,6 @@ async def login(
flow_value = FlowType(flow).value if flow else None
scope_list = [s.strip() for s in scopes.split(",")] if scopes else None
if force and not ctx_obj.json_output and not ctx_obj.quiet:
ctx_obj.echo("Warning: Forcing login will overwrite any existing connection.", color="yellow")
if not ctx_obj.json_output:
ctx_obj.echo(f"Starting login for {provider}...", color="cyan")
try:
session_info = await actx.runtime_client.start_login(
provider=provider,
@@ -273,7 +141,6 @@ async def login(
base_url=base_url,
force=force,
)
session_id = session_info["id"]
status = session_info.get("status")
login_result = {"status": "started", "record_status": status}
@@ -285,9 +152,6 @@ async def login(
if action_type == "open_url":
auth_url = next_action["url"]
if not ctx_obj.json_output and not ctx_obj.quiet:
ctx_obj.echo("Opening browser to continue login...", color="cyan")
ctx_obj.echo(f"Visit: {auth_url}", color="cyan")
import webbrowser
try:
@@ -295,13 +159,6 @@ async def login(
except Exception:
pass
if not ctx_obj.json_output and not ctx_obj.quiet:
ctx_obj.echo(
"\nLogin process started. The connection will be updated automatically once complete.",
color="yellow",
)
ctx_obj.echo(f"Session ID: {session_id}")
logger.info(
"client_event event=login provider={} connection={} flow={} status={}",
provider,
@@ -310,31 +167,16 @@ async def login(
login_result["status"],
)
except Exception:
if not ctx_obj.json_output:
logger.warning("client_event event=login provider={} connection={} status=failure", provider, connection)
raise
if ctx_obj.json_output:
ctx_obj.print_json(
{
"status": login_result.get("status", "success"),
"provider": provider,
"connection": connection,
"record_status": login_result.get("record_status"),
}
)
elif login_result.get("status") == "success":
ctx_obj.echo(
f"Already logged in to {provider} ({connection}). Use the --force flag to overwrite and open the browser.",
color="green",
)
elif login_result.get("status") == "started":
ctx_obj.echo(
f"Login started for {provider} ({connection}). Run 'authsome provider list' to verify completion.",
color="green",
)
else:
ctx_obj.echo(f"Successfully logged in to {provider} ({connection}).", color="green")
ctx_obj.print_json(
{
"status": login_result.get("status", "success"),
"provider": provider,
"connection": connection,
"record_status": login_result.get("record_status"),
}
)
@cli.command(name="scan")
@@ -457,27 +299,15 @@ async def scan(ctx_obj: ContextObj, connection: str, auto_import: bool) -> None:
item["env_var"],
)
if ctx_obj.json_output:
ctx_obj.print_json(
{
"connection": connection,
"import": should_import,
"configured_count": len(configured),
"imported_count": imported,
"results": results,
}
)
else:
if not results:
ctx_obj.echo("No API key providers found to process.", color="yellow")
else:
for item in results:
env_hint = f" ({item['env_var']})" if item.get("env_var") else ""
source_hint = f" from {item['source']}" if item.get("source") else ""
ctx_obj.echo(f"{item['provider']}: {item['status']}{env_hint}{source_hint}")
if configured and not should_import:
ctx_obj.echo("Import skipped by user.", color="yellow")
ctx_obj.echo(f"Imported {imported} provider(s).", color="green")
ctx_obj.print_json(
{
"connection": connection,
"import": should_import,
"configured_count": len(configured),
"imported_count": imported,
"results": results,
}
)
@cli.command()
@@ -490,10 +320,7 @@ async def logout(ctx_obj: ContextObj, provider: str, connection: str) -> None:
await actx.runtime_client.logout(provider, connection)
logger.info("client_event event=logout provider={} connection={}", provider, connection)
if ctx_obj.json_output:
ctx_obj.print_json({"status": "logged_out", "provider": provider, "connection": connection})
else:
ctx_obj.echo(f"Logged out of {provider} ({connection}).", color="green")
ctx_obj.print_json({"status": "logged_out", "provider": provider, "connection": connection})
@connections.command(name="set-default")
@@ -504,10 +331,7 @@ async def set_default_connection(ctx_obj: ContextObj, provider: str, connection:
"""Set the default CONNECTION for PROVIDER."""
actx = await ctx_obj.initialize()
await actx.runtime_client.set_default_connection(provider, connection)
if ctx_obj.json_output:
ctx_obj.print_json({"status": "ok", "provider": provider, "default_connection": connection})
else:
ctx_obj.echo(f"Default connection for {provider} set to {connection}.", color="green")
ctx_obj.print_json({"status": "ok", "provider": provider, "default_connection": connection})
@provider.command()
@@ -519,10 +343,7 @@ async def revoke(ctx_obj: ContextObj, provider: str) -> None:
await actx.runtime_client.revoke(provider)
logger.info("client_event event=revoke provider={} connection=all", provider)
if ctx_obj.json_output:
ctx_obj.print_json({"status": "revoked", "provider": provider})
else:
ctx_obj.echo(f"Revoked all credentials for {provider}.", color="green")
ctx_obj.print_json({"status": "revoked", "provider": provider})
@provider.command()
@@ -534,10 +355,7 @@ async def remove(ctx_obj: ContextObj, provider: str) -> None:
await actx.runtime_client.remove(provider)
logger.info("client_event event=remove provider={} connection=all", provider)
if ctx_obj.json_output:
ctx_obj.print_json({"status": "removed", "provider": provider})
else:
ctx_obj.echo(f"Removed provider {provider}.", color="green")
ctx_obj.print_json({"status": "removed", "provider": provider})
@connections.command(name="inspect")
@@ -585,11 +403,8 @@ async def inspect_provider(ctx_obj: ContextObj, provider: str) -> None:
data["connections"] = provider_group["connections"]
break
if ctx_obj.json_output:
data.pop("schema_version", None)
ctx_obj.print_json(data)
else:
ctx_obj.echo(json_lib.dumps(data, indent=2))
data.pop("schema_version", None)
ctx_obj.print_json(data)
@cli.command(context_settings=dict(ignore_unknown_options=True))
@@ -617,32 +432,15 @@ async def register(ctx_obj: ContextObj, path: str, force: bool, yes: bool) -> No
sys.exit(1)
try:
data = json_lib.loads(filepath.read_text(encoding="utf-8"))
data = json.loads(filepath.read_text(encoding="utf-8"))
definition = ProviderDefinition.model_validate(data)
endpoints_to_check = _validate_provider_endpoints(definition)
if not ctx_obj.json_output and not ctx_obj.quiet and not yes and not force:
ctx_obj.echo(f"Registering '{definition.name}' provider:")
for name, val, _ in endpoints_to_check:
ctx_obj.echo(f" - {name}: {val}")
if definition.oauth and definition.oauth.token_url:
prompt_msg = f"Register '{definition.name}' with token endpoint {definition.oauth.token_url}? [y/N]"
elif definition.api_url:
prompt_msg = f"Register '{definition.name}' with host {definition.api_url}? [y/N]"
else:
prompt_msg = f"Register '{definition.name}' provider? [y/N]"
if not click.confirm(prompt_msg, default=False):
ctx_obj.echo("Registration aborted.", color="yellow")
sys.exit(0)
await actx.runtime_client.register_provider(definition.model_dump(mode="json"), force=force)
endpoints = [ep for _, ep, _ in endpoints_to_check]
logger.info("client_event event=register provider={} endpoints={}", definition.name, endpoints)
ctx_obj.print_json({"status": "registered", "provider": definition.name})
warnings = []
for name, val, is_host in endpoints_to_check:
@@ -653,16 +451,12 @@ async def register(ctx_obj: ContextObj, path: str, force: bool, yes: bool) -> No
if is_host and "://" not in target:
target = f"https://{target}"
if not ctx_obj.quiet:
ctx_obj.echo(f"Testing reachability for {name}...", color="cyan")
try:
requests.head(target, timeout=5, allow_redirects=True)
except requests.RequestException as e:
warnings.append(f"{name} ({val}) is unreachable: {e}")
if warnings and not ctx_obj.quiet:
for warning in warnings:
ctx_obj.echo(f"Warning: {warning}", color="yellow")
ctx_obj.print_json({"status": "registered", "provider": definition.name, "warnings": warnings})
except Exception as exc:
ctx_obj.print_json({"error": exc.__class__.__name__, "message": f"Failed to register provider: {exc}"})
sys.exit(format_error_code(exc))
@@ -691,13 +485,7 @@ async def init(ctx_obj: ContextObj) -> None:
"effective_encryption_source": whoami_data.get("effective_encryption_source"),
"encryption_backend": whoami_data.get("encryption_backend"),
}
if ctx_obj.json_output:
ctx_obj.print_json(data)
else:
ctx_obj.echo(f"Initialized authsome at {home}", color="green")
ctx_obj.echo(f"Profile: {identity.handle}")
ctx_obj.echo(f"DID: {identity.did}")
ctx_obj.echo(f"Master Key Source: {_render_encryption_backend(whoami_data)}")
ctx_obj.print_json(data)
@cli.group(name="profile")
@@ -723,12 +511,7 @@ async def profile_create(ctx_obj: ContextObj, handle: str | None) -> None:
"registration_status": "registered" if identity_meta.registered else "local",
"switched": True,
}
if ctx_obj.json_output:
ctx_obj.print_json(data)
else:
ctx_obj.echo(f"Created local profile {identity_meta.handle}", color="green")
ctx_obj.echo("Switched to new profile")
ctx_obj.echo(f"DID: {identity_meta.did}")
ctx_obj.print_json(data)
@profile.command(name="use")
@@ -748,11 +531,7 @@ async def profile_use(ctx_obj: ContextObj, handle: str) -> None:
"profile": identity_meta.handle,
"did": identity_meta.did,
}
if ctx_obj.json_output:
ctx_obj.print_json(data)
else:
ctx_obj.echo(f"Active profile: {data['profile']}", color="green")
ctx_obj.echo(f"DID: {data['did']}")
ctx_obj.print_json(data)
@cli.command()
@@ -804,36 +583,7 @@ async def whoami(ctx_obj: ContextObj) -> None:
"issues": issues,
}
if ctx_obj.json_output:
ctx_obj.print_json(data)
else:
ctx_obj.echo(f"Authsome Version: {data['authsome_version']}")
ctx_obj.echo(f"Home Directory: {data['home_directory']}")
ctx_obj.echo(f"Profile: {data['profile']}")
if data["principal_id"]:
ctx_obj.echo(f"Principal: {data['principal_id']}")
if data["vault_id"]:
ctx_obj.echo(f"Vault: {data['vault_id']}")
if data["did"]:
ctx_obj.echo(f"DID: {data['did']}")
if data["registration_status"]:
ctx_obj.echo(f"Registration: {data['registration_status']}")
ctx_obj.echo(f"Daemon URL: {data['daemon_url']}")
status_color = "green" if vault_status == "OK" else "red"
ctx_obj.echo(f"Encryption: {_render_encryption_backend(data)} [", nl=False)
ctx_obj.echo(vault_status, color=status_color, nl=False)
ctx_obj.echo("]")
if issues:
ctx_obj.echo("\nIssues:", color="red")
for issue in issues:
ctx_obj.echo(f" - {issue}", color="red")
ctx_obj.echo(f"\nConnected Providers: {data['connected_providers_count']}")
if connected_providers:
for p in sorted(connected_providers, key=lambda x: x["name"]):
suffix = "connection" if p["count"] == 1 else "connections"
ctx_obj.echo(f" {p['name']} ({p['count']} {suffix})")
ctx_obj.print_json(data)
@cli.command()
@@ -844,29 +594,9 @@ async def doctor(ctx_obj: ContextObj) -> None:
results = await actx.doctor()
all_ok = results.get("status") == "ready"
if ctx_obj.json_output:
ctx_obj.print_json(results)
if not all_ok:
sys.exit(1)
else:
for key, val in results.get("checks", {}).items():
ok = val == "ok"
ctx_obj.emit(f"{key}: ", nl=False)
ctx_obj.emit("OK" if ok else "FAIL", color="green" if ok else "red")
issues = results.get("issues", [])
if issues:
ctx_obj.echo("\nIssues found:", color="red")
for issue in issues:
ctx_obj.echo(f" - {issue}", color="red")
warnings = results.get("warnings", [])
if warnings:
ctx_obj.echo("\nWarnings:", color="yellow")
for warning in warnings:
ctx_obj.echo(f" - {warning}", color="yellow")
if not all_ok:
sys.exit(1)
ctx_obj.print_json(results)
if not all_ok:
sys.exit(1)
if __name__ == "__main__":
src/authsome/server/routes/health.py
+1 -21
View File
@@ -2,11 +2,9 @@
from __future__ import annotations
import asyncio
import secrets
from typing import Literal, cast
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi import APIRouter, Depends, Request
from authsome import __version__
from authsome.server.credential_service import AuthService
@@ -111,24 +109,6 @@ async def ready(
)
_rekey_lock = asyncio.Lock()
@router.post("/rekey")
async def rekey(
request: Request,
auth: AuthService = Depends(get_protected_auth_service),
) -> dict[str, str]:
_ = request
async with _rekey_lock:
new_key_bytes = secrets.token_bytes(32)
try:
await auth.vault.rekey(new_key_bytes)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return {"status": "ok", "message": "Master key successfully rotated"}
@router.get("/whoami")
async def whoami(
request: Request,
src/authsome/vault/__init__.py
+2 -48
View File
@@ -5,13 +5,10 @@ from __future__ import annotations
import builtins
import json
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
from key_value.aio._utils.compound import uncompound_key
from loguru import logger
from typing import TYPE_CHECKING
from authsome.store.interfaces import AppStore
from authsome.vault.crypto import VaultCrypto, create_crypto, create_rekey_crypto
from authsome.vault.crypto import VaultCrypto, create_crypto
if TYPE_CHECKING:
from authsome.vault.crypto import VaultCrypto
@@ -110,49 +107,6 @@ class Vault:
_ = identity
return await self._app_store.check_integrity()
async def rekey(self, new_key_bytes: bytes) -> None:
"""Re-encrypt all encrypted keys in the underlying KV store using a new master key."""
old_crypto = self.crypto
old_crypto.assert_rekey_supported()
# 1. Create a new crypto instance using the new key
new_crypto = create_rekey_crypto(new_key_bytes)
# 2. Iterate over all entries in the underlying DiskStore cache
# DiskStore stores compound keys as collection::key
cache = cast(Any, self._app_store.kv)._cache
# Collect all keys first to avoid iterating while modifying
all_compound_keys = list(cache.iterkeys())
# Perform in-place re-encryption
reencrypted_count = 0
for comp_key in all_compound_keys:
try:
collection, key = uncompound_key(comp_key)
except Exception:
continue
if collection == "config" or key == "__index__":
continue
# Retrieve and decrypt the entry
val = await self._app_store.kv.get(key, collection=collection)
if val is not None and "data" in val:
ciphertext = val["data"]
# Decrypt with old crypto and re-encrypt with new crypto
plaintext = old_crypto.decrypt(ciphertext)
new_ciphertext = new_crypto.encrypt(plaintext)
# Store the re-encrypted value back
await self._app_store.kv.put(key, {"data": new_ciphertext}, collection=collection)
reencrypted_count += 1
# 3. Delegate persistence to the active backend
old_crypto.persist_rekeyed_key(new_key_bytes)
# 4. Clear the active crypto in memory so it reloads on next access
self._crypto = None
logger.info("Rekey completed successfully. Re-encrypted {} keys.", reencrypted_count)
async def close(self) -> None:
"""Release resources."""
await self._app_store.close()
src/authsome/vault/crypto.py
-92
View File
@@ -54,16 +54,6 @@ class VaultCrypto(ABC):
"""Decrypt a compact ciphertext string and return plaintext."""
...
@abstractmethod
def persist_rekeyed_key(self, new_key_bytes: bytes) -> None:
"""Persist a newly rotated master key for this backend."""
...
@abstractmethod
def assert_rekey_supported(self) -> None:
"""Raise when this backend cannot perform an in-place rekey."""
...
def _encode(nonce: bytes, ct_with_tag: bytes) -> str:
"""Pack nonce + ciphertext+tag into a single dot-separated base64 string."""
@@ -141,27 +131,6 @@ class LocalFileCrypto(_AesGcmCrypto):
logger.info("Generated new master key at {}", self._key_file)
return master_key
def persist_rekeyed_key(self, new_key_bytes: bytes) -> None:
"""Atomically replace the local master key file after a rekey."""
_validate_master_key_bytes(new_key_bytes)
temp_path = self._key_file.with_suffix(".tmp")
key_data = {
"version": 1,
"key": base64.b64encode(new_key_bytes).decode("ascii"),
"algorithm": "AES-256-GCM",
"note": "Local master key for authsome. Protect this file.",
}
temp_path.write_text(json.dumps(key_data, indent=2), encoding="utf-8")
try:
os.chmod(temp_path, 0o600)
except OSError:
pass
temp_path.replace(self._key_file)
def assert_rekey_supported(self) -> None:
"""Local file storage supports in-place rekey."""
return None
class KeyringCrypto(_AesGcmCrypto):
"""AES-256-GCM with master key stored in the OS keyring."""
@@ -183,26 +152,6 @@ class KeyringCrypto(_AesGcmCrypto):
raise EncryptionUnavailableError("OS keyring is unavailable and no master key could be created.")
return master_key
def persist_rekeyed_key(self, new_key_bytes: bytes) -> None:
"""Store a rotated master key in the OS keyring."""
_validate_master_key_bytes(new_key_bytes)
key_b64_str = base64.b64encode(new_key_bytes).decode("ascii")
try:
import keyring as kr
except ImportError as exc:
raise RuntimeError(
"The 'keyring' package is required for keyring mode. Install it with: pip install keyring"
) from exc
try:
kr.set_password(_KEYRING_SERVICE, _KEYRING_USERNAME, key_b64_str)
except Exception as exc:
raise RuntimeError(f"Failed to store new master key in OS keyring: {exc}") from exc
def assert_rekey_supported(self) -> None:
"""OS keyring storage supports in-place rekey."""
return None
class EnvVarCrypto(_AesGcmCrypto):
"""AES-256-GCM with master key supplied via AUTHSOME_MASTER_KEY."""
@@ -226,30 +175,12 @@ class EnvVarCrypto(_AesGcmCrypto):
raise EncryptionUnavailableError(f"{_MASTER_KEY_ENV_VAR} is set but empty.")
return _decode_master_key(raw_value.strip(), _MASTER_KEY_ENV_VAR)
def persist_rekeyed_key(self, new_key_bytes: bytes) -> None:
"""Reject in-place rekey for externally supplied master keys."""
_ = new_key_bytes
self.assert_rekey_supported()
def assert_rekey_supported(self) -> None:
"""Reject rekey for externally managed master keys."""
raise ValueError(
"Vault rekey is unavailable while using AUTHSOME_MASTER_KEY. "
"Update the external master key and migrate data from a writable backend first."
)
def _new_master_key() -> bytes:
"""Generate a new 256-bit master key."""
return secrets.token_bytes(_KEY_SIZE_BYTES)
def _validate_master_key_bytes(master_key: bytes) -> None:
"""Validate already-decoded master key bytes."""
if len(master_key) != _KEY_SIZE_BYTES:
raise EncryptionUnavailableError(f"Master key must be {_KEY_SIZE_BYTES} bytes; got {len(master_key)} bytes.")
def _decode_master_key(encoded_value: str, source: str) -> bytes:
"""Decode and validate a base64-encoded master key."""
try:
@@ -329,29 +260,6 @@ def _create_auto_crypto(key_file: Path | None) -> VaultCrypto:
return LocalFileCrypto(key_file)
def create_rekey_crypto(new_key_bytes: bytes) -> VaultCrypto:
"""Create an ephemeral in-memory backend for vault re-encryption."""
class _RekeyCrypto(_AesGcmCrypto):
@property
def source_id(self) -> str:
return "rekey"
@property
def source_description(self) -> str:
return "In-memory rekey backend"
def persist_rekeyed_key(self, new_key_bytes: bytes) -> None:
_ = new_key_bytes
raise RuntimeError("In-memory rekey backend cannot persist master keys")
def assert_rekey_supported(self) -> None:
"""This helper backend is never used as a persisted store."""
raise RuntimeError("In-memory rekey backend cannot validate persisted rekey support")
return _RekeyCrypto(new_key_bytes)
def create_crypto(key_file: Path | None, mode: str = "auto") -> VaultCrypto:
"""Factory: return the appropriate VaultCrypto backend for the given mode."""
if mode == "auto":
tests/server/test_pop_auth.py
-33
View File
@@ -89,39 +89,6 @@ def test_health_and_ready_report_encryption_details(monkeypatch, tmp_path: Path)
assert "AUTHSOME_MASTER_KEY" in ready_response.json()["encryption_backend"]
def test_rekey_rotates_local_vault(monkeypatch, tmp_path: Path) -> None:
monkeypatch.setenv("AUTHSOME_HOME", str(tmp_path))
monkeypatch.delenv("AUTHSOME_DEPLOYMENT_MODE", raising=False)
identity = create_identity(tmp_path, "steady-wisely-boldly-0042")
with TestClient(create_app()) as client:
client.post("/identities/register", json={"handle": identity.handle, "did": identity.did})
resolved = asyncio.run(client.app.state.ownership_resolver.resolve(identity=identity.handle))
asyncio.run(client.app.state.vault.put("key1", "secret-value-1", collection=f"vault:{resolved.vault_id}"))
response = client.post("/rekey", json={}, headers=_auth_header(tmp_path, "POST", "/rekey", b"{}"))
assert response.status_code == 200
assert response.json()["status"] == "ok"
assert (
asyncio.run(client.app.state.vault.get("key1", collection=f"vault:{resolved.vault_id}")) == "secret-value-1"
)
def test_rekey_rejects_env_master_key(monkeypatch, tmp_path: Path) -> None:
monkeypatch.setenv("AUTHSOME_HOME", str(tmp_path))
monkeypatch.setenv("AUTHSOME_MASTER_KEY", base64.b64encode(b"\x03" * 32).decode("ascii"))
monkeypatch.delenv("AUTHSOME_DEPLOYMENT_MODE", raising=False)
identity = create_identity(tmp_path, "steady-wisely-boldly-0042")
with TestClient(create_app()) as client:
client.post("/identities/register", json={"handle": identity.handle, "did": identity.did})
response = client.post("/rekey", json={}, headers=_auth_header(tmp_path, "POST", "/rekey", b"{}"))
assert response.status_code == 400
assert "AUTHSOME_MASTER_KEY" in response.json()["detail"]
def test_hosted_registration_requires_claim(monkeypatch, tmp_path: Path) -> None:
monkeypatch.setenv("AUTHSOME_HOME", str(tmp_path))
monkeypatch.setenv("AUTHSOME_DEPLOYMENT_MODE", "hosted")
tests/vault/test_rekey.py
-167
View File
@@ -1,167 +0,0 @@
"""Tests for the vault rekey / master key rotation functionality."""
from __future__ import annotations
import base64
import json
import secrets
from pathlib import Path
import pytest
from authsome.store.local import LocalAppStore
from authsome.vault import Vault
from authsome.vault.crypto import LocalFileCrypto
@pytest.mark.asyncio
class TestVaultRekey:
"""Vault rekeying tests."""
async def test_rekey_local_key_mode(self, tmp_path: Path) -> None:
# 1. Initialize store and vault in local_key mode
app_store = LocalAppStore(tmp_path)
await app_store.ensure_initialized()
master_key_path = tmp_path / "server" / "master.key"
vault = Vault(
app_store=app_store,
crypto_mode="local_key",
master_key_path=master_key_path,
)
# 2. Write some encrypted secrets into different collections
await vault.put("key1", "secret-value-1", collection="col1")
await vault.put("key2", "secret-value-2", collection="col1")
await vault.put("key3", "secret-value-3", collection="col2")
# 3. Read them back to verify they are decrypted correctly
assert await vault.get("key1", collection="col1") == "secret-value-1"
assert await vault.get("key2", collection="col1") == "secret-value-2"
assert await vault.get("key3", collection="col2") == "secret-value-3"
# Capture old key bytes
old_key_data = json.loads(master_key_path.read_text(encoding="utf-8"))
old_key_bytes = base64.b64decode(old_key_data["key"])
# 4. Generate a new key and perform rekeying
new_key_bytes = secrets.token_bytes(32)
await vault.rekey(new_key_bytes)
# 5. Read them back with the rekeyed vault
assert await vault.get("key1", collection="col1") == "secret-value-1"
assert await vault.get("key2", collection="col1") == "secret-value-2"
assert await vault.get("key3", collection="col2") == "secret-value-3"
# 6. Verify that the key file is updated
new_key_data = json.loads(master_key_path.read_text(encoding="utf-8"))
updated_key_bytes = base64.b64decode(new_key_data["key"])
assert updated_key_bytes == new_key_bytes
assert updated_key_bytes != old_key_bytes
# 7. Verify that trying to decrypt with old key directly fails
# Get raw ciphertext from store
raw_val = await app_store.kv.get("key1", collection="col1")
assert raw_val is not None
ciphertext = raw_val["data"]
old_key_path = tmp_path / "server" / "old-master.key"
old_key_path.write_text(
json.dumps(
{
"version": 1,
"key": base64.b64encode(old_key_bytes).decode("ascii"),
"algorithm": "AES-256-GCM",
"note": "Local master key for authsome. Protect this file.",
}
),
encoding="utf-8",
)
old_crypto = LocalFileCrypto(old_key_path)
from authsome.errors import EncryptionUnavailableError
with pytest.raises(EncryptionUnavailableError):
old_crypto.decrypt(ciphertext)
# But new crypto decrypts it successfully!
new_key_path = tmp_path / "server" / "new-master.key"
new_key_path.write_text(
json.dumps(
{
"version": 1,
"key": base64.b64encode(new_key_bytes).decode("ascii"),
"algorithm": "AES-256-GCM",
"note": "Local master key for authsome. Protect this file.",
}
),
encoding="utf-8",
)
new_crypto = LocalFileCrypto(new_key_path)
assert new_crypto.decrypt(ciphertext) == "secret-value-1"
await app_store.close()
async def test_rekey_rejects_env_mode(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
app_store = LocalAppStore(tmp_path)
await app_store.ensure_initialized()
monkeypatch.setenv("AUTHSOME_MASTER_KEY", base64.b64encode(secrets.token_bytes(32)).decode("ascii"))
vault = Vault(app_store=app_store, crypto_mode="env")
await vault.put("key1", "secret-value-1", collection="col1")
with pytest.raises(ValueError, match="AUTHSOME_MASTER_KEY"):
await vault.rekey(secrets.token_bytes(32))
assert await vault.get("key1", collection="col1") == "secret-value-1"
await app_store.close()
async def test_rekey_keyring_mode(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
# Mock keyring set/get methods
keyring_store: dict[str, str] = {}
class MockKeyring:
@staticmethod
def set_password(service: str, username: str, password: str) -> None:
keyring_store[f"{service}:{username}"] = password
@staticmethod
def get_password(service: str, username: str) -> str | None:
return keyring_store.get(f"{service}:{username}")
import sys
sys.modules["keyring"] = MockKeyring # type: ignore
# 1. Initialize store and vault in keyring mode
app_store = LocalAppStore(tmp_path)
await app_store.ensure_initialized()
# Seed initial key in keyring
initial_key = secrets.token_bytes(32)
MockKeyring.set_password(
"authsome",
"master_key",
base64.b64encode(initial_key).decode("ascii"),
)
vault = Vault(
app_store=app_store,
crypto_mode="keyring",
)
# 2. Write some encrypted secrets
await vault.put("key1", "keyring-secret", collection="col1")
assert await vault.get("key1", collection="col1") == "keyring-secret"
# 3. Rekey the vault
new_key_bytes = secrets.token_bytes(32)
await vault.rekey(new_key_bytes)
# 4. Verify keyring is updated and vault can still read the secret
stored_b64 = MockKeyring.get_password("authsome", "master_key")
assert stored_b64 is not None
assert base64.b64decode(stored_b64) == new_key_bytes
assert await vault.get("key1", collection="col1") == "keyring-secret"
await app_store.close()