Add agent security controls, broker auth, and capability discovery

This commit is contained in:
thePR0M3TH3AN committed 2026-02-28 22:32:38 -05:00
1 parent 1febbd4b2c
commit fe96ccd7c7
15 files changed
+2381 -99

No files matched your search

+50 -2
View File
@@ -22,6 +22,13 @@ import bcrypt
from seedpass.core.manager import PasswordManager
from seedpass.core.entry_types import EntryType
from seedpass.core.agent_export_policy import (
allowed_kinds,
filter_index_for_allowed_kinds,
full_export_allowed,
kind_export_allowed,
load_export_policy,
)
from seedpass.core.api import UtilityService
_RATE_LIMIT = int(os.getenv("SEEDPASS_RATE_LIMIT", "100"))
@@ -138,18 +145,24 @@ def _reload_relays(request: Request, relays: list[str]) -> None:
logger.error("Failed to initialize NostrClient with relays %s: %s", relays, exc)
def start_server(fingerprint: str | None = None) -> str:
def start_server(
fingerprint: str | None = None, unlock_password: str | None = None
) -> str:
"""Initialize global state and return a random API token.
Parameters
----------
fingerprint:
Optional seed profile fingerprint to select before starting the server.
unlock_password:
Optional master password used to unlock vault during startup.
"""
if fingerprint is None:
pm = PasswordManager()
else:
pm = PasswordManager(fingerprint=fingerprint)
if unlock_password is not None:
pm.unlock_vault(unlock_password)
app.state.pm = pm
raw_token = secrets.token_urlsafe(32)
app.state.token_hash = bcrypt.hashpw(raw_token.encode(), bcrypt.gensalt())
@@ -184,6 +197,10 @@ def _validate_encryption_path(request: Request, path: Path) -> Path:
raise HTTPException(status_code=400, detail=str(e))
def _header_truthy(value: str | None) -> bool:
return str(value or "").strip().lower() in {"1", "true", "yes", "on"}
@app.get("/api/v1/entry")
async def search_entry(
request: Request, query: str, authorization: str | None = Header(None)
@@ -524,11 +541,19 @@ def export_totp(
request: Request,
authorization: str | None = Header(None),
password: str | None = Header(None, alias="X-SeedPass-Password"),
agent_profile: str | None = Header(None, alias="X-SeedPass-Agent-Profile"),
) -> dict:
"""Return all stored TOTP entries in JSON format."""
_check_token(request, authorization)
_require_password(request, password)
_require_unlocked(request)
if _header_truthy(agent_profile):
policy = load_export_policy()
if not kind_export_allowed(policy, EntryType.TOTP.value):
raise HTTPException(
status_code=403,
detail="Policy denied TOTP export for agent profile",
)
pm = _get_pm(request)
key = getattr(pm, "KEY_TOTP_DET", None) or getattr(pm, "parent_seed", None)
return pm.entry_manager.export_totp_entries(key)
@@ -696,12 +721,31 @@ def export_vault(
request: Request,
authorization: str | None = Header(None),
password: str | None = Header(None, alias="X-SeedPass-Password"),
agent_profile: str | None = Header(None, alias="X-SeedPass-Agent-Profile"),
policy_filtered: str | None = Header(None, alias="X-SeedPass-Policy-Filtered"),
):
"""Export the vault and return the encrypted file."""
_check_token(request, authorization)
_require_password(request, password)
_require_unlocked(request)
is_agent = _header_truthy(agent_profile)
is_filtered = _header_truthy(policy_filtered)
pm = _get_pm(request)
if is_agent:
policy = load_export_policy()
if is_filtered:
index_data = pm.vault.load_index()
filtered = filter_index_for_allowed_kinds(index_data, allowed_kinds(policy))
payload = json.dumps(
filtered, sort_keys=True, separators=(",", ":")
).encode("utf-8")
data = pm.vault.encryption_manager.encrypt_data(payload)
return Response(content=data, media_type="application/octet-stream")
if not full_export_allowed(policy):
raise HTTPException(
status_code=403,
detail="Policy denied full vault export for agent profile",
)
path = pm.handle_export_database()
if path is None:
raise HTTPException(status_code=500, detail="Export failed")
@@ -865,7 +909,11 @@ def unlock_vault(
_record_unlock_failure(request, raw_token)
raise HTTPException(status_code=401, detail="Invalid password")
duration = pm.unlock_vault(password)
return {"status": "unlocked", "duration": float(duration)}
return {
"status": "unlocked",
"duration": float(duration),
"help_hint": "Use /docs or run `seedpass --help` for command and endpoint discovery.",
}
@app.post("/api/v1/shutdown")
+2
View File
@@ -39,6 +39,7 @@ deterministic_totp_option = typer.Option(
# Sub command groups
from . import entry, vault, nostr, config, fingerprint, util, api, agent
from .capabilities import register_capabilities_command
app.add_typer(entry.app, name="entry")
app.add_typer(vault.app, name="vault")
@@ -48,6 +49,7 @@ app.add_typer(fingerprint.app, name="fingerprint")
app.add_typer(util.app, name="util")
app.add_typer(api.app, name="api")
app.add_typer(agent.app, name="agent")
register_capabilities_command(app)
def run() -> None:
+1056 -88
View File
File diff suppressed because it is too large. Load diff
+53 -2
View File
@@ -1,17 +1,68 @@
from __future__ import annotations
from typing import Optional
import click
import typer
import uvicorn
from .. import api as api_module
from seedpass.core.auth_broker import AuthBrokerError, resolve_password as resolve_broker_password
app = typer.Typer(help="Run the API server")
@app.command("start")
def api_start(ctx: typer.Context, host: str = "127.0.0.1", port: int = 8000) -> None:
def api_start(
ctx: typer.Context,
host: str = "127.0.0.1",
port: int = 8000,
unlock: bool = typer.Option(
False, "--unlock", help="Unlock vault during startup using the selected auth broker"
),
auth_broker: str = typer.Option(
"prompt",
"--auth-broker",
help="Password source when --unlock is used (prompt|env|keyring|command)",
click_type=click.Choice(["prompt", "env", "keyring", "command"], case_sensitive=False),
),
password_env: str = typer.Option(
"SEEDPASS_PASSWORD", "--password-env", help="Env var used when --auth-broker=env"
),
broker_service: str = typer.Option(
"seedpass",
"--broker-service",
help="Keyring service name when --auth-broker=keyring",
),
broker_account: Optional[str] = typer.Option(
None,
"--broker-account",
help="Keyring account (defaults to active fingerprint when available)",
),
broker_command: Optional[str] = typer.Option(
None,
"--broker-command",
help="Command that prints password to stdout for --auth-broker=command",
),
) -> None:
"""Start the SeedPass API server."""
token = api_module.start_server(ctx.obj.get("fingerprint"))
unlock_password: str | None = None
if unlock:
if auth_broker.lower() == "prompt":
unlock_password = typer.prompt("Master password", hide_input=True)
else:
account = broker_account or str((ctx.obj or {}).get("fingerprint") or "default")
try:
unlock_password = resolve_broker_password(
broker=auth_broker,
password_env=password_env,
broker_service=broker_service,
broker_account=account,
broker_command=broker_command,
)
except AuthBrokerError as exc:
raise typer.BadParameter(str(exc)) from exc
token = api_module.start_server(ctx.obj.get("fingerprint"), unlock_password)
typer.echo(
f"API token: {token}\nWARNING: Store this token securely; it cannot be recovered."
)
+117
View File
@@ -0,0 +1,117 @@
from __future__ import annotations
import json
from typing import Any
import click
import typer
def build_capabilities() -> dict[str, Any]:
"""Return a deterministic capability map for CLI/API discovery."""
return {
"schema_version": 1,
"product": "SeedPass",
"interfaces": {
"cli": {
"root_commands": [
"lock",
"gui",
"capabilities",
"entry",
"vault",
"nostr",
"config",
"fingerprint",
"util",
"api",
"agent",
],
"global_options": [
"--fingerprint",
"--no-clipboard",
"--deterministic-totp",
],
},
"api": {
"discovery": ["/docs"],
"auth_headers": ["Authorization: Bearer <token>"],
"sensitive_auth_headers": ["X-SeedPass-Password"],
},
},
"security_features": {
"auth_brokers": {
"available": ["env", "keyring", "command", "prompt"],
"agent_commands": ["agent init", "agent get"],
"vault_commands": ["vault unlock"],
"api_commands": ["api start --unlock"],
},
"policy": {
"commands": ["agent policy-show", "agent policy-lint", "agent policy-set"],
"default_effect": "deny",
},
"tokens": {
"commands": ["agent token-issue", "agent token-list", "agent token-revoke"],
"supports_ttl": True,
"supports_use_limits": True,
"supports_revocation": True,
},
"redaction": {
"default_safe_output": True,
"reveal_override": "agent get --reveal",
},
"audit": {
"log_file": "~/.seedpass/agent_audit.log",
"integrity": "HMAC chained append-only records",
},
},
"high_risk_operations": [
"vault export",
"vault reveal-parent-seed",
"entry get (sensitive kinds)",
"api /api/v1/vault/export",
],
"help_hints": [
"Run `seedpass --help` for top-level commands.",
"Run `seedpass <group> --help` for detailed options.",
"Use `seedpass capabilities --format json` for machine-readable discovery.",
],
}
def emit_capabilities(output_format: str) -> str:
data = build_capabilities()
if output_format == "json":
return json.dumps(data, indent=2, sort_keys=True)
lines: list[str] = []
lines.append("SeedPass Capabilities (schema v1)")
lines.append("")
lines.append("CLI root commands:")
lines.append(", ".join(data["interfaces"]["cli"]["root_commands"]))
lines.append("")
lines.append("Security highlights:")
lines.append("- Auth brokers: env, keyring, command, prompt")
lines.append("- Policy controls: agent policy-show/policy-lint/policy-set")
lines.append("- Scoped tokens: issue/list/revoke with TTL and use limits")
lines.append("- Safe output redaction default with explicit --reveal")
lines.append("- Audited access log with chained HMAC signatures")
lines.append("")
lines.append("Discovery hints:")
for hint in data["help_hints"]:
lines.append(f"- {hint}")
return "\n".join(lines)
def register_capabilities_command(app: typer.Typer) -> None:
@app.command("capabilities")
def capabilities_command(
format: str = typer.Option(
"text",
"--format",
help="Output format for capabilities map",
click_type=click.Choice(["text", "json"], case_sensitive=False),
)
) -> None:
"""Show a deterministic capabilities map for users and agents."""
typer.echo(emit_capabilities(format.lower()))
+13 -1
View File
@@ -9,6 +9,7 @@ import typer
import click
from .common import _get_entry_service, EntryType
from seedpass.core.agent_export_policy import kind_export_allowed, load_export_policy
from seedpass.core.entry_types import ALL_ENTRY_TYPES
from utils.clipboard import ClipboardUnavailableError
@@ -357,9 +358,20 @@ def entry_totp_codes(ctx: typer.Context) -> None:
@app.command("export-totp")
def entry_export_totp(
ctx: typer.Context, file: str = typer.Option(..., help="Output file")
ctx: typer.Context,
file: str = typer.Option(..., help="Output file"),
agent_profile: bool = typer.Option(
False,
"--agent-profile",
help="Apply agent export policy controls",
),
) -> None:
"""Export all TOTP secrets to a JSON file."""
if agent_profile:
policy = load_export_policy()
if not kind_export_allowed(policy, EntryType.TOTP.value):
typer.echo("Error: Policy denied TOTP export for agent profile.")
raise typer.Exit(code=1)
service = _get_entry_service(ctx)
data = service.export_totp_entries()
Path(file).write_text(json.dumps(data, indent=2))
+88 -3
View File
@@ -4,6 +4,7 @@ import json
from pathlib import Path
from typing import Optional
import click
import typer
from .common import (
@@ -12,16 +13,57 @@ from .common import (
UnlockRequest,
BackupParentSeedRequest,
)
from seedpass.core.agent_export_policy import (
allowed_kinds,
filter_index_for_allowed_kinds,
full_export_allowed,
load_export_policy,
)
from seedpass.core.auth_broker import AuthBrokerError, resolve_password as resolve_broker_password
app = typer.Typer(help="Manage the entire vault")
@app.command("export")
def vault_export(
ctx: typer.Context, file: str = typer.Option(..., help="Output file")
ctx: typer.Context,
file: str = typer.Option(..., help="Output file"),
agent_profile: bool = typer.Option(
False,
"--agent-profile",
help="Apply agent export policy controls",
),
policy_filtered: bool = typer.Option(
False,
"--policy-filtered",
help="Export only policy-allowed subset of entries",
),
) -> None:
"""Export the vault profile to an encrypted file."""
vault_service, _profile, _sync = _get_services(ctx)
if agent_profile:
policy = load_export_policy()
if policy_filtered:
manager = getattr(vault_service, "_manager", None)
vault = getattr(manager, "vault", None)
enc_mgr = getattr(vault, "encryption_manager", None)
if vault is None or enc_mgr is None:
typer.echo("Error: policy-filtered export unavailable in current context")
raise typer.Exit(code=1)
index_data = vault.load_index()
filtered = filter_index_for_allowed_kinds(index_data, allowed_kinds(policy))
payload = json.dumps(
filtered, sort_keys=True, separators=(",", ":")
).encode("utf-8")
data = enc_mgr.encrypt_data(payload)
Path(file).write_bytes(data)
typer.echo(str(file))
return
if not full_export_allowed(policy):
typer.echo(
"Error: Policy denied full vault export for agent profile. Use --policy-filtered."
)
raise typer.Exit(code=1)
data = vault_service.export_profile()
Path(file).write_bytes(data)
typer.echo(str(file))
@@ -55,16 +97,59 @@ def vault_change_password(ctx: typer.Context) -> None:
@app.command("unlock")
def vault_unlock(ctx: typer.Context) -> None:
def vault_unlock(
ctx: typer.Context,
auth_broker: str = typer.Option(
"prompt",
"--auth-broker",
help="Password source for unlock (prompt|env|keyring|command)",
click_type=click.Choice(["prompt", "env", "keyring", "command"], case_sensitive=False),
),
password_env: str = typer.Option(
"SEEDPASS_PASSWORD", "--password-env", help="Env var used when --auth-broker=env"
),
broker_service: str = typer.Option(
"seedpass",
"--broker-service",
help="Keyring service name when --auth-broker=keyring",
),
broker_account: Optional[str] = typer.Option(
None,
"--broker-account",
help="Keyring account (defaults to active fingerprint when available)",
),
broker_command: Optional[str] = typer.Option(
None,
"--broker-command",
help="Command that prints password to stdout for --auth-broker=command",
),
) -> None:
"""Unlock the vault for the active profile."""
vault_service, _profile, _sync = _get_services(ctx)
password = typer.prompt("Master password", hide_input=True)
password: str
if auth_broker.lower() == "prompt":
password = typer.prompt("Master password", hide_input=True)
else:
account = broker_account or str((ctx.obj or {}).get("fingerprint") or "default")
try:
password = resolve_broker_password(
broker=auth_broker,
password_env=password_env,
broker_service=broker_service,
broker_account=account,
broker_command=broker_command,
)
except AuthBrokerError as exc:
raise typer.BadParameter(str(exc)) from exc
try:
resp = vault_service.unlock(UnlockRequest(password=password))
except Exception as exc: # pragma: no cover - pass through errors
typer.echo(f"Error: {exc}")
raise typer.Exit(code=1)
typer.echo(f"Unlocked in {resp.duration:.2f}s")
typer.echo(
"Tip: run `seedpass --help` (and `seedpass <command> --help`) to discover available features."
)
@app.command("lock")
+86
View File
@@ -0,0 +1,86 @@
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from constants import APP_DIR
from seedpass.core.entry_types import ALL_ENTRY_TYPES, EntryType
DEFAULT_EXPORT_POLICY: dict[str, Any] = {
"allow_kinds": [
EntryType.PASSWORD.value,
EntryType.TOTP.value,
EntryType.KEY_VALUE.value,
],
"allow_export_import": False,
"export": {"allow_full_vault": False},
}
def _policy_path() -> Path:
return APP_DIR / "agent_policy.json"
def load_export_policy() -> dict[str, Any]:
path = _policy_path()
if not path.exists():
return dict(DEFAULT_EXPORT_POLICY)
try:
data = json.loads(path.read_text(encoding="utf-8"))
except Exception:
return dict(DEFAULT_EXPORT_POLICY)
if not isinstance(data, dict):
return dict(DEFAULT_EXPORT_POLICY)
policy = dict(DEFAULT_EXPORT_POLICY)
allow_kinds = data.get("allow_kinds")
if isinstance(allow_kinds, list):
policy["allow_kinds"] = [
str(v) for v in allow_kinds if str(v) in ALL_ENTRY_TYPES
]
if isinstance(data.get("allow_export_import"), bool):
policy["allow_export_import"] = data["allow_export_import"]
export_cfg = data.get("export")
if isinstance(export_cfg, dict):
policy["export"] = {
"allow_full_vault": bool(export_cfg.get("allow_full_vault", False))
}
elif isinstance(data.get("allow_export_import"), bool):
policy["export"] = {"allow_full_vault": bool(data["allow_export_import"])}
return policy
def full_export_allowed(policy: dict[str, Any]) -> bool:
return bool(
policy.get("allow_export_import", False)
or policy.get("export", {}).get("allow_full_vault", False)
)
def allowed_kinds(policy: dict[str, Any]) -> set[str]:
kinds = policy.get("allow_kinds", [])
if not isinstance(kinds, list):
return set()
return {str(v) for v in kinds if str(v) in ALL_ENTRY_TYPES}
def kind_export_allowed(policy: dict[str, Any], kind: str) -> bool:
return kind in allowed_kinds(policy)
def filter_index_for_allowed_kinds(index_data: dict[str, Any], kinds: set[str]) -> dict[str, Any]:
entries = index_data.get("entries", {})
if not isinstance(entries, dict):
entries = {}
filtered_entries = {}
for idx, entry in entries.items():
if not isinstance(entry, dict):
continue
kind = str(entry.get("kind", entry.get("type", "")))
if kind in kinds:
filtered_entries[str(idx)] = entry
return {
"schema_version": int(index_data.get("schema_version", 1)),
"entries": filtered_entries,
}
+73
View File
@@ -0,0 +1,73 @@
from __future__ import annotations
import os
import shlex
import subprocess
class AuthBrokerError(ValueError):
"""Raised when non-interactive password retrieval fails."""
def _password_from_env(var_name: str) -> str:
value = os.getenv(var_name)
if not value:
raise AuthBrokerError(
f"Missing password env var '{var_name}'. Export it before running this command."
)
return value
def _password_from_keyring(service: str, account: str) -> str:
try:
import keyring # type: ignore
except Exception as exc:
raise AuthBrokerError(
"Keyring broker requested but 'keyring' package is not installed."
) from exc
value = keyring.get_password(service, account)
if not value:
raise AuthBrokerError(
f"No secret found in keyring for service='{service}' account='{account}'."
)
return value
def _password_from_command(command: str) -> str:
if not command.strip():
raise AuthBrokerError("Command broker requires a non-empty --broker-command.")
try:
proc = subprocess.run(
shlex.split(command),
capture_output=True,
text=True,
check=True,
)
except subprocess.CalledProcessError as exc:
stderr = (exc.stderr or "").strip()
detail = f": {stderr}" if stderr else ""
raise AuthBrokerError(f"Command broker failed{detail}") from exc
value = (proc.stdout or "").strip()
if not value:
raise AuthBrokerError("Command broker returned empty output.")
return value
def resolve_password(
*,
broker: str,
password_env: str = "SEEDPASS_PASSWORD",
broker_service: str = "seedpass",
broker_account: str = "default",
broker_command: str | None = None,
) -> str:
"""Resolve password using a non-interactive broker strategy."""
mode = broker.strip().lower()
if mode == "env":
return _password_from_env(password_env)
if mode == "keyring":
return _password_from_keyring(broker_service, broker_account)
if mode == "command":
return _password_from_command(broker_command or "")
raise AuthBrokerError(f"Unsupported auth broker '{broker}'.")
+38
View File
@@ -5,6 +5,7 @@ import base64
import pytest
from seedpass import api
import seedpass.core.agent_export_policy as export_policy
import string
from seedpass.core.password_generation import PasswordGenerator, PasswordPolicy
from seedpass.core.encryption import EncryptionManager
@@ -462,6 +463,7 @@ async def test_vault_unlock_endpoint(client):
res = await cl.post("/api/v1/vault/unlock", headers=headers)
assert res.status_code == 200
assert res.json()["status"] == "unlocked"
assert "help_hint" in res.json()
assert called["password"] == "pw"
@@ -610,6 +612,42 @@ async def test_vault_export_endpoint(client, tmp_path):
assert res.status_code == 401
@pytest.mark.anyio
async def test_vault_export_endpoint_denied_for_agent_profile(
client, tmp_path, monkeypatch
):
cl, token = client
monkeypatch.setattr(export_policy, "APP_DIR", tmp_path)
headers = {
"Authorization": f"Bearer {token}",
"X-SeedPass-Password": "pw",
"X-SeedPass-Agent-Profile": "true",
}
res = await cl.post("/api/v1/vault/export", headers=headers)
assert res.status_code == 403
assert "Policy denied full vault export" in res.json()["detail"]
@pytest.mark.anyio
async def test_totp_export_endpoint_denied_for_agent_profile(
client, tmp_path, monkeypatch
):
cl, token = client
monkeypatch.setattr(export_policy, "APP_DIR", tmp_path)
(tmp_path / "agent_policy.json").write_text(
'{"allow_kinds":["password"],"allow_export_import":false}',
encoding="utf-8",
)
headers = {
"Authorization": f"Bearer {token}",
"X-SeedPass-Password": "pw",
"X-SeedPass-Agent-Profile": "true",
}
res = await cl.get("/api/v1/totp/export", headers=headers)
assert res.status_code == 403
assert "Policy denied TOTP export" in res.json()["detail"]
@pytest.mark.anyio
async def test_backup_parent_seed_endpoint(client, tmp_path):
cl, token = client
+304 -3
View File
@@ -84,6 +84,36 @@ def test_agent_init_requires_password_env(monkeypatch, tmp_path):
assert "Missing password env var" in combined
def test_agent_init_keyring_broker(monkeypatch, tmp_path):
monkeypatch.setattr(agent_cli, "APP_DIR", tmp_path)
calls = {}
def fake_resolve_password(**kwargs):
calls.update(kwargs)
return "AgentPass123!"
monkeypatch.setattr(agent_cli, "resolve_broker_password", fake_resolve_password)
result = runner.invoke(
app,
[
"agent",
"init",
"--seed",
"abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about",
"--auth-broker",
"keyring",
"--broker-service",
"seedpass",
"--broker-account",
"ci-agent",
],
)
assert result.exit_code == 0
assert calls["broker"] == "keyring"
assert calls["broker_service"] == "seedpass"
assert calls["broker_account"] == "ci-agent"
def test_agent_get_requires_fingerprint(monkeypatch):
monkeypatch.setenv("SEEDPASS_PASSWORD", "pw")
result = runner.invoke(app, ["agent", "get", "example"])
@@ -120,7 +150,9 @@ def test_agent_get_password_json(monkeypatch):
lambda **kwargs: {"allow_kinds": ["password"], "deny_private_reveal": []},
)
result = runner.invoke(app, ["--fingerprint", "ABC123", "agent", "get", "example"])
result = runner.invoke(
app, ["--fingerprint", "ABC123", "agent", "get", "example", "--reveal"]
)
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["status"] == "ok"
@@ -163,7 +195,7 @@ def test_agent_get_denied_by_policy(monkeypatch):
assert result.exit_code == 1
payload = json.loads(result.stdout)
assert payload["status"] == "denied"
assert payload["reason"] in {"kind_not_allowed", "private_kind_blocked"}
assert payload["reason"].startswith("policy_deny:")
def test_agent_get_ambiguous_returns_error_payload(monkeypatch):
@@ -228,7 +260,9 @@ def test_agent_get_key_value_json(monkeypatch):
lambda **kwargs: {"allow_kinds": ["key_value"], "deny_private_reveal": []},
)
result = runner.invoke(app, ["--fingerprint", "ABC123", "agent", "get", "key"])
result = runner.invoke(
app, ["--fingerprint", "ABC123", "agent", "get", "key", "--reveal"]
)
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["status"] == "ok"
@@ -236,6 +270,147 @@ def test_agent_get_key_value_json(monkeypatch):
assert payload["secret"] == "v123"
def test_agent_get_safe_output_masks_by_default(monkeypatch):
monkeypatch.setenv("SEEDPASS_PASSWORD", "pw")
monkeypatch.setattr(
agent_cli,
"PasswordManager",
lambda fingerprint, password: SimpleNamespace(
fingerprint=fingerprint, password=password
),
)
class DummyEntryService:
def __init__(self, _pm):
pass
def search_entries(self, _q):
return [(2, "Example", "", "", False, EntryType.PASSWORD)]
def retrieve_entry(self, _i):
return {"type": EntryType.PASSWORD.value, "label": "Example", "length": 12}
def generate_password(self, _length, _index):
return "secret-pass"
monkeypatch.setattr(agent_cli, "EntryService", DummyEntryService)
monkeypatch.setattr(
agent_cli,
"_load_policy",
lambda **kwargs: {
"version": 1,
"default_effect": "allow",
"rules": [],
"output": {"safe_output_default": True},
"allow_kinds": ["password"],
"deny_private_reveal": [],
},
)
result = runner.invoke(app, ["--fingerprint", "ABC123", "agent", "get", "example"])
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["status"] == "ok"
assert payload["safe_output"] is True
assert payload["secret"] != "secret-pass"
def test_agent_token_issue_and_use_limit(monkeypatch, tmp_path):
monkeypatch.setattr(agent_cli, "APP_DIR", tmp_path)
monkeypatch.setenv("SEEDPASS_PASSWORD", "pw")
monkeypatch.setattr(
agent_cli,
"PasswordManager",
lambda fingerprint, password: SimpleNamespace(
fingerprint=fingerprint, password=password
),
)
class DummyEntryService:
def __init__(self, _pm):
pass
def search_entries(self, _q):
return [(2, "Example", "", "", False, EntryType.PASSWORD)]
def retrieve_entry(self, _i):
return {"type": EntryType.PASSWORD.value, "label": "Example", "length": 12}
def generate_password(self, _length, _index):
return "secret-pass"
monkeypatch.setattr(agent_cli, "EntryService", DummyEntryService)
monkeypatch.setattr(
agent_cli,
"_load_policy",
lambda **kwargs: {
"version": 1,
"default_effect": "allow",
"rules": [],
"output": {"safe_output_default": False},
"allow_kinds": ["password"],
"deny_private_reveal": [],
},
)
issue = runner.invoke(
app,
[
"agent",
"token-issue",
"--name",
"ci-agent",
"--scope",
"read",
"--kind",
"password",
"--uses",
"1",
"--ttl",
"600",
],
)
assert issue.exit_code == 0
issued_payload = json.loads(issue.stdout)
token = issued_payload["token"]
first = runner.invoke(
app,
[
"--fingerprint",
"ABC123",
"agent",
"get",
"example",
"--token",
token,
"--reveal",
],
)
assert first.exit_code == 0
first_payload = json.loads(first.stdout)
assert first_payload["status"] == "ok"
assert first_payload["token_uses_remaining"] == 0
second = runner.invoke(
app,
[
"--fingerprint",
"ABC123",
"agent",
"get",
"example",
"--token",
token,
"--reveal",
],
)
assert second.exit_code == 1
second_payload = json.loads(second.stdout)
assert second_payload["status"] == "denied"
assert second_payload["reason"] == "token_exhausted"
def test_agent_get_unsupported_kind_payload(monkeypatch):
monkeypatch.setenv("SEEDPASS_PASSWORD", "pw")
monkeypatch.setattr(
@@ -288,3 +463,129 @@ def test_agent_get_denied_when_policy_invalid(monkeypatch, tmp_path):
payload = json.loads(result.stdout)
assert payload["status"] == "denied"
assert payload["reason"] == "invalid_policy"
def test_agent_get_keyring_defaults_account_to_fingerprint(monkeypatch):
calls = {}
monkeypatch.setattr(
agent_cli,
"resolve_broker_password",
lambda **kwargs: calls.update(kwargs) or "pw",
)
monkeypatch.setattr(
agent_cli,
"PasswordManager",
lambda fingerprint, password: SimpleNamespace(
fingerprint=fingerprint, password=password
),
)
class DummyEntryService:
def __init__(self, _pm):
pass
def search_entries(self, _q):
return [(2, "Example", "", "", False, EntryType.PASSWORD)]
def retrieve_entry(self, _i):
return {"type": EntryType.PASSWORD.value, "label": "Example", "length": 12}
def generate_password(self, _length, _index):
return "secret-pass"
monkeypatch.setattr(agent_cli, "EntryService", DummyEntryService)
monkeypatch.setattr(
agent_cli,
"_load_policy",
lambda **kwargs: {
"version": 1,
"default_effect": "allow",
"rules": [],
"output": {"safe_output_default": False},
"allow_kinds": ["password"],
"deny_private_reveal": [],
},
)
result = runner.invoke(
app,
[
"--fingerprint",
"ABC123",
"agent",
"get",
"example",
"--auth-broker",
"keyring",
"--broker-service",
"seedpass",
"--reveal",
],
)
assert result.exit_code == 0
assert calls["broker"] == "keyring"
assert calls["broker_account"] == "ABC123"
def test_agent_bootstrap_context(monkeypatch, tmp_path):
monkeypatch.setattr(agent_cli, "APP_DIR", tmp_path)
result = runner.invoke(app, ["--fingerprint", "ABC123", "agent", "bootstrap-context"])
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["status"] == "ok"
assert payload["fingerprint"] == "ABC123"
assert payload["policy"]["status"] == "ok"
assert "supported" in payload["auth_brokers"]
assert "agent token-issue" in payload["commands"]["tokens"]
def test_agent_bootstrap_context_invalid_policy(monkeypatch, tmp_path):
monkeypatch.setattr(agent_cli, "APP_DIR", tmp_path)
(tmp_path / "agent_policy.json").write_text("{", encoding="utf-8")
result = runner.invoke(app, ["agent", "bootstrap-context"])
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["policy"]["status"] == "invalid"
def test_agent_posture_check_clean(monkeypatch, tmp_path):
monkeypatch.setattr(agent_cli, "APP_DIR", tmp_path)
result = runner.invoke(app, ["agent", "posture-check"])
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["status"] == "ok"
assert payload["finding_count"] == 0
assert payload["highest_severity"] == "info"
def test_agent_posture_check_invalid_policy_fails(monkeypatch, tmp_path):
monkeypatch.setattr(agent_cli, "APP_DIR", tmp_path)
(tmp_path / "agent_policy.json").write_text("{", encoding="utf-8")
result = runner.invoke(app, ["agent", "posture-check", "--fail-on", "critical"])
assert result.exit_code == 1
payload = json.loads(result.stdout)
assert payload["policy_status"] == "invalid"
assert payload["highest_severity"] == "critical"
assert any(f["id"] == "policy_invalid" for f in payload["findings"])
def test_agent_posture_check_high_risk_policy(monkeypatch, tmp_path):
monkeypatch.setattr(agent_cli, "APP_DIR", tmp_path)
policy = {
"version": 1,
"default_effect": "allow",
"rules": [],
"output": {"safe_output_default": False, "redact_fields": ["secret"]},
"export": {"allow_full_vault": True},
"allow_export_import": True,
"allow_kinds": ["password"],
"deny_private_reveal": [],
}
(tmp_path / "agent_policy.json").write_text(json.dumps(policy), encoding="utf-8")
result = runner.invoke(app, ["agent", "posture-check", "--fail-on", "high"])
assert result.exit_code == 1
payload = json.loads(result.stdout)
ids = {f["id"] for f in payload["findings"]}
assert "policy_default_allow" in ids
assert "safe_output_disabled" in ids
assert "export_import_allowed" in ids
+26
View File
@@ -24,9 +24,35 @@ def test_cli_vault_unlock(monkeypatch):
result = runner.invoke(app, ["vault", "unlock"])
assert result.exit_code == 0
assert "Unlocked in" in result.stdout
assert "seedpass --help" in result.stdout
assert called["pw"] == "pw"
def test_cli_vault_unlock_with_env_broker(monkeypatch):
called = {}
broker_called = {}
def unlock_vault(pw):
called["pw"] = pw
return 0.25
pm = SimpleNamespace(unlock_vault=unlock_vault, select_fingerprint=lambda fp: None)
monkeypatch.setattr(cli_common, "PasswordManager", lambda: pm)
monkeypatch.setattr(
"seedpass.cli.vault.resolve_broker_password",
lambda **kwargs: broker_called.update(kwargs) or "broker-pw",
)
monkeypatch.setattr(
cli.typer,
"prompt",
lambda *a, **k: (_ for _ in ()).throw(AssertionError("prompt should not be called")),
)
result = runner.invoke(app, ["vault", "unlock", "--auth-broker", "env"])
assert result.exit_code == 0
assert called["pw"] == "broker-pw"
assert broker_called["broker"] == "env"
def test_cli_entry_add_search_sync(monkeypatch):
calls = {}
+86
View File
@@ -11,6 +11,7 @@ from seedpass.cli import common as cli_common
from seedpass.cli import api as cli_api
from seedpass import cli
from seedpass.core.entry_types import EntryType
import seedpass.core.agent_export_policy as export_policy
runner = CliRunner()
@@ -85,6 +86,18 @@ def test_vault_export(monkeypatch, tmp_path):
assert out_path.read_bytes() == b"data"
def test_vault_export_denied_for_agent_profile_by_default(monkeypatch, tmp_path):
monkeypatch.setattr(export_policy, "APP_DIR", tmp_path)
monkeypatch.setattr(cli_common.VaultService, "export_profile", lambda self: b"data")
monkeypatch.setattr(cli_common, "PasswordManager", lambda: SimpleNamespace())
out_path = tmp_path / "out.json"
result = runner.invoke(
app, ["vault", "export", "--file", str(out_path), "--agent-profile"]
)
assert result.exit_code == 1
assert "Policy denied full vault export" in result.stdout
def test_vault_import(monkeypatch, tmp_path):
called = {}
@@ -386,6 +399,24 @@ def test_generate_password(monkeypatch):
assert "secretpw" in result.stdout
def test_capabilities_text():
result = runner.invoke(app, ["capabilities"])
assert result.exit_code == 0
assert "SeedPass Capabilities" in result.stdout
assert "Auth brokers" in result.stdout
def test_capabilities_json():
result = runner.invoke(app, ["capabilities", "--format", "json"])
assert result.exit_code == 0
import json
payload = json.loads(result.stdout)
assert payload["schema_version"] == 1
assert "cli" in payload["interfaces"]
assert "agent" in payload["interfaces"]["cli"]["root_commands"]
def test_api_start_passes_fingerprint(monkeypatch):
"""Ensure the API start command forwards the selected fingerprint."""
called = {}
@@ -402,6 +433,41 @@ def test_api_start_passes_fingerprint(monkeypatch):
assert called.get("fp") == "abc"
def test_api_start_unlock_uses_env_broker(monkeypatch):
called = {}
broker_called = {}
def fake_start(fingerprint=None, unlock_password=None):
called["fp"] = fingerprint
called["pw"] = unlock_password
return "tok"
monkeypatch.setattr(cli_api.api_module, "start_server", fake_start)
monkeypatch.setattr(cli_api, "uvicorn", SimpleNamespace(run=lambda *a, **k: None))
monkeypatch.setattr(
cli_api,
"resolve_broker_password",
lambda **kwargs: broker_called.update(kwargs) or "broker-pw",
)
result = runner.invoke(
app,
[
"--fingerprint",
"abc",
"api",
"start",
"--unlock",
"--auth-broker",
"env",
],
)
assert result.exit_code == 0
assert called.get("fp") == "abc"
assert called.get("pw") == "broker-pw"
assert broker_called.get("broker") == "env"
def test_entry_list_passes_fingerprint(monkeypatch):
"""Ensure entry commands receive the fingerprint."""
called = {}
@@ -564,6 +630,26 @@ def test_entry_export_totp(monkeypatch, tmp_path):
assert called.get("called") is True
def test_entry_export_totp_denied_for_agent_profile(monkeypatch, tmp_path):
monkeypatch.setattr(export_policy, "APP_DIR", tmp_path)
(tmp_path / "agent_policy.json").write_text(
'{"allow_kinds":["password"],"allow_export_import":false}',
encoding="utf-8",
)
pm = SimpleNamespace(
entry_manager=SimpleNamespace(export_totp_entries=lambda seed: {"entries": []}),
parent_seed="seed",
select_fingerprint=lambda fp: None,
)
monkeypatch.setattr(cli_common, "PasswordManager", lambda: pm)
out = tmp_path / "t.json"
result = runner.invoke(
app, ["entry", "export-totp", "--file", str(out), "--agent-profile"]
)
assert result.exit_code == 1
assert "Policy denied TOTP export" in result.stdout
def test_entry_totp_codes(monkeypatch):
called = {}