diff --git a/artifacts/todos.txt b/artifacts/todos.txt index 59ed6089..8675575e 100644 --- a/artifacts/todos.txt +++ b/artifacts/todos.txt @@ -1,4 +1 @@ -TODO.md:1:# SeedPass TODO -memory-updates/legacy-2026-02-26T13-32-55Z.md:21:- Logged explicit remaining tasks to finish checklist item #8 in both `docs/supply_chain_release_integrity.md` and `TODO.md`. -src/todo/TODO_2026-02-26T02-43-51Z.md:7:# TODO src/utils/memory_protection.py:6:# TODO: Replace this Python implementation with a Rust/WASM module for diff --git a/docs/agent_autonomy_security_plan.md b/docs/agent_autonomy_security_plan.md new file mode 100644 index 00000000..b8946fb2 --- /dev/null +++ b/docs/agent_autonomy_security_plan.md @@ -0,0 +1,280 @@ +# SeedPass Agent Autonomy Security Plan + +## Goal +Implement a staged security architecture that enables autonomous agents to use SeedPass safely in production environments without relying on plaintext secrets or overly broad access. + +## Program Structure +- Phase A: Foundation + Add policy engine hardening, brokered auth, scoped tokens, redaction, and auditable access. +- Phase B: Guardrails + Add approval gates, high-risk secret isolation, and deterministic export controls. +- Phase C: Operations + Add posture tooling, rotation workflows, recovery hardening, and conflict-safe sync semantics. + +## 1. Native Non-Interactive Auth Brokers +- Problem: + Agent auth depends heavily on env vars and manual password flow. +- Implementation: + Add an auth broker abstraction with providers: + `env` (backward-compatible), `keyring` (OS-native), `command` (for HSM/YubiKey helpers). + Add broker options to `seedpass agent init/get` and then `vault unlock`, `api start`. + Add broker plugins for `pkcs11` and `yubikey` as optional modules later. +- Data/Config: + `--auth-broker`, `--broker-service`, `--broker-account`, `--broker-command`. +- Security: + Never persist broker-returned secrets; memory-only use. +- Tests: + broker resolution, failure handling, per-profile account defaults. +- Acceptance: + agent workflows can authenticate without password env vars by using keyring/broker commands. + +## 2. Fine-Grained RBAC Per Profile +- Problem: + Coarse kind-level permissions are insufficient. +- Implementation: + Extend policy rules to enforce allow/deny by: + entry path, label regex, kind, field, operation (`read/create/update/export`). + Enforce in CLI agent commands, API endpoints, and service layer wrappers. +- Data/Config: + `agent_policy.json` schema v2 with explicit operation matrix and precedence. +- Tests: + policy precedence, deny overrides, path+field restrictions. +- Acceptance: + deterministic policy decisions for every operation and entry field. + +## 3. Short-Lived Scoped Tokens +- Problem: + Long-lived static tokens increase blast radius. +- Implementation: + Expand token store to support: + scope lists, TTL, usage count, profile binding, optional IP/host binding, revocation reason. + Add token introspection endpoint/command. +- Data/Config: + `agent_tokens.json` versioned schema. +- Tests: + expiry, revoke, scope denial, kind denial, reuse after exhaustion. +- Acceptance: + all agent access can use revocable TTL-bound scoped credentials. + +## 4. One-Time Secret Leases +- Problem: + Secrets may be exposed repeatedly after retrieval. +- Implementation: + Add lease records tied to token and resource: + fetch-once or N-use, automatic invalidation, lease audit linkage. + Optional server-side one-time retrieval IDs for API consumers. +- Tests: + first-use success, second-use denial, race behavior under concurrency. +- Acceptance: + one-time leases are enforced reliably under parallel access. + +## 5. Policy-Enforced Redaction +- Problem: + Secrets can leak through stdout/logging by default. +- Implementation: + Make safe output mode default in agent/API. + Redact by field policy with optional reveal override requiring permission. + Add structured redaction markers. +- Tests: + output snapshot tests for masked/unmasked modes. +- Acceptance: + secrets are masked by default in automation logs. + +## 6. Auditable Event Log +- Problem: + Access traceability is weak without tamper evidence. +- Implementation: + Keep append-only signed chain log. + Add verifier command: + `seedpass agent audit-verify`. + Add export to JSONL with integrity proofs. +- Tests: + chain continuity, tamper detection, clock skew handling. +- Acceptance: + any log tampering is detectable; access events are attributable. + +## 7. Approval Gates +- Problem: + High-risk actions need step-up control. +- Implementation: + Define gate policies for `export`, `reveal_parent_seed`, `private_key_retrieval`. + Add approval plugins: + local second factor, signed approval token, webhook approver. + Add TTL and single-use approval grants. +- Tests: + denied without approval, success with valid grant, grant expiry. +- Acceptance: + risky operations cannot execute without valid step-up approval. + +## 8. Secret-Class Isolation +- Problem: + High-risk materials should not share unlock context with standard secrets. +- Implementation: + Partition vault by class: + standard vs high-risk (`seed`, `ssh`, `pgp`). + Separate derived encryption keys and unlock factors. + Per-class lock/unlock state. +- Data/Config: + class partition metadata in encrypted config. +- Tests: + class lock boundary, cross-class access denial. +- Acceptance: + compromise of standard unlock path does not unlock high-risk partitions. + +## 9. Agent Identities +- Problem: + Tokens alone do not model lifecycle and ownership. +- Implementation: + Add first-class agent accounts: + id, owner, policy binding, rotation interval, revocation state. + Tokens must be issued under agent identity. +- Data/Config: + `agent_identities.json` versioned schema. +- Tests: + identity disable/revoke, policy inheritance. +- Acceptance: + every token and access event maps to a managed agent identity. + +## 10. Safer Automation Primitives +- Problem: + cron jobs often use plaintext env secrets. +- Implementation: + Add job runner integration: + broker-only auth, scoped policy binding, ephemeral runtime context. + Add template commands for systemd timers and cron wrappers. +- Tests: + non-interactive scheduled task success without plaintext secret files. +- Acceptance: + supported scheduled workflows avoid plaintext env password dependence. + +## 11. Recovery Hardening +- Problem: + Recovery process can be brittle or under-tested. +- Implementation: + Add Shamir split/recovery workflows for parent seed protection. + Add scheduled backup verification drills with signed reports. +- Tests: + split/recombine determinism, threshold enforcement, drill report generation. +- Acceptance: + enforceable recovery posture with verifiable drills. + +## 12. Conflict-Safe Sync Primitives +- Problem: + multi-device edits can conflict unpredictably. +- Implementation: + Introduce deterministic merge strategy: + monotonic timestamps + deterministic tie-breaker + field-level merge rules. + Encrypt metadata needed for conflict resolution. +- Tests: + concurrent edit replay, deterministic final state across nodes. +- Acceptance: + same inputs always produce same merged vault state. + +## 13. Policy as Code +- Problem: + policy lifecycle lacks strong review workflows. +- Implementation: + Add schema validation, linting, canonical formatting, diff tooling, policy tests. + Add optional signed policy bundles and hash pinning. +- Tests: + lint failures, schema migration tests, canonical output tests. +- Acceptance: + policy changes are reviewable, testable, and reproducible. + +## 14. Deterministic Export Controls +- Problem: + full-vault exports are dangerous for agent profiles. +- Implementation: + Enforce policy-restricted subset exports: + allowed kinds/paths/fields only, default deny full-vault for agent identities. + Add deterministic manifest for exported subset. +- Tests: + blocked full export, allowed subset export correctness. +- Acceptance: + agent profiles cannot perform unrestricted exports by default. + +## 15. Security Posture Tooling +- Problem: + drift and misconfiguration can silently increase risk. +- Implementation: + Add `seedpass agent posture-check`: + stale credentials, weak broker config, overbroad policy rules, missing rotation, gate bypasses. + Output machine-readable report and severity levels. +- Tests: + posture check fixtures with expected findings. +- Acceptance: + actionable posture report with CI-friendly nonzero exit on critical findings. + +## Recommended Delivery Order +1. Item 1 completion across all auth entry points. +2. Items 2, 7, 14 enforcement expansion. +3. Items 3, 4, 9 identity-token lifecycle. +4. Items 6, 13, 15 governance and audit maturity. +5. Items 8, 10, 11, 12 deeper operational hardening. + +## Definition of Done (Cross-Cutting) +- No plaintext secret persistence introduced. +- New schemas are versioned and migration tested. +- All new controls covered by unit + integration tests. +- CLI/API paths return deterministic, machine-readable error payloads. + +## Help and Discoverability TODO Backlog +### Objective +Make SeedPass self-describing for autonomous agents and human operators so feature discovery does not depend on external docs or code spelunking. + +### TODO 1: Login-Time Guidance +- Add concise post-unlock guidance text in: + `seedpass vault unlock` and API `/api/v1/vault/unlock`. +- Include links/hints to: + `seedpass --help`, `seedpass --help`, API `/docs`. +- Add tests verifying hint visibility and stability. + +### TODO 2: Machine-Readable Capability Surface +- Add `seedpass capabilities --format json|yaml|text`. +- Include: + command groups, commands, option schema, auth modes, output mode (`json`/text), risk class, and policy gates. +- Ensure deterministic ordering for diffability in CI. + +### TODO 3: Command Metadata Registry +- Add a central metadata registry for CLI/API capabilities. +- Fields: + `name`, `summary`, `requires_unlock`, `supports_noninteractive`, `requires_approval`, `sensitive_outputs`, `policy_operations`. +- Generate help and capabilities output from the same registry to prevent drift. + +### TODO 4: Agent-Focused Help Profile +- Add `--help-profile agent|human`. +- Agent profile: + concise machine-actionable semantics, failure modes, deterministic examples, JSON schemas. +- Human profile: + richer narrative hints and workflow examples. + +### TODO 5: Output Schema Discovery +- Add `--output json` and `--describe-output` support to critical commands (`agent get`, `vault unlock`, token/policy commands). +- Publish expected JSON field schema in command help and capability output. + +### TODO 6: Risk and Approval Annotations in Help +- Mark risky commands in help with a stable tag: + `risk:high`, `approval:required`. +- Apply to: + export, reveal-parent-seed, private key retrieval paths. + +### TODO 7: Context Bootstrap Command for Agents +- Add `seedpass agent bootstrap-context`. +- Return: + active profile fingerprint, policy status, broker options available, token status summary, feature support map, version info. + +### TODO 8: API Discoverability Parity +- Ensure API endpoint metadata mirrors CLI capability data. +- Add a protected endpoint: + `/api/v1/capabilities`. +- Keep OpenAPI tags aligned with CLI command groups. + +### TODO 9: Regression Tests for Help Contracts +- Snapshot tests for: + root help, group help, key command help, capabilities JSON. +- Add contract tests ensuring new commands include metadata fields. + +### TODO 10: Documentation Sync Automation +- Add a CI check that compares: + generated capabilities output vs docs command reference. +- Fail CI when command metadata and docs diverge. diff --git a/memory-update.md b/memory-update.md index 748f771a..a8b8edd9 100644 --- a/memory-update.md +++ b/memory-update.md @@ -4,3 +4,112 @@ * The default branch is `beta`. `git log` should target `beta` instead of `main` when reading history for this repository. * A `CHANGELOG.md` file exists and we append changes. Since no existing release notes dir was found, I created `releases/draft-YYYYMMDD.md`. * We use `poetry run black .` for formatting tests properly, but I reverted the changes to `torch/scripts` formatting since they were not meant to be changed by this agent. + +## Key findings +- The `scheduler-flow.md` script execution failed due to a missing runner (`codex`). +- The automated `bug-reproducer-agent` prompt execution failed. +- Fallback manual investigation identified one active issue: `BIP85 Non-Standard Derivation Path` (using `app_no=2` instead of standard). This is documented in `KNOWN_ISSUES.md`. + +## Patterns / reusable knowledge +- Future runs should verify the runner environment variables (`SCHEDULER_PROMPT_PATH`, `SCHEDULER_AGENT`, `SCHEDULER_CADENCE`) are set correctly before invoking `run-selected-prompt.mjs`. +- If the runner fails, manual fallback or alternative automation (e.g., Python script) is required. + +## Warnings / gotchas +- `torch/scripts/agent/run-selected-prompt.mjs` depends on an external runner (`codex` or similar) which may not be available in all environments. + +# Memory Update — agent-security-foundation — 2026-03-01 + +## Key findings +- `src/seedpass/cli/agent.py` previously used a flat policy (`allow_kinds` / `deny_private_reveal`) with no rule-level operation, regex, or field matching. +- Introducing a versioned rule model can remain backward-compatible by auto-normalizing legacy policy files into structured `rules`. +- Empty `kinds` in deny rules must not be generated unintentionally for legacy migration, otherwise deny rules become wildcard blocks. + +## Patterns / reusable knowledge +- For agent automation hardening, implement in this order for minimal breakage: policy normalization, enforcement hooks, safe-output redaction, token scoping/TTL/use limits, then append-only audit logging. +- Keep compatibility fields (`allow_kinds`, `deny_private_reveal`, `allow_export_import`) in the normalized policy output so older tooling still functions. +- Tests in `src/tests/test_cli_agent_mode.py` are the right place to cover agent policy regressions and token lease semantics. + +## Warnings / gotchas +- CI environment may not have global `pytest`; use `./.venv/bin/pytest` in this repo. +- Agent command tests that monkeypatch `_load_policy` with legacy payloads require in-command normalization to avoid false policy denials. + +# Memory Update — auth-broker-phase1 — 2026-03-01 + +## Key findings +- A dedicated broker abstraction in `seedpass/core/auth_broker.py` allows non-interactive secret retrieval without coupling broker logic into CLI command modules. +- Legacy env-var auth can remain default while enabling keyring/command brokers via explicit CLI options. +- For per-profile non-interactive unlock, defaulting keyring account to fingerprint in `agent get` avoids account collisions across profiles. + +## Patterns / reusable knowledge +- Introduce broker modes in this order: `env` (compat), `keyring` (native OS storage), `command` (bridge to HSM/YubiKey helper tooling). +- Keep broker-specific failures converted to `typer.BadParameter` so automation receives consistent deterministic exit behavior. +- Agent-mode tests are the fastest place to validate broker migration without touching broader interactive flows. + +## Warnings / gotchas +- `keyring` package is optional in current dependency set; keyring broker must fail with a clear install/error message when unavailable. +- Command broker output must be strictly stdout-only password text; any wrapper script should avoid additional logs on stdout. + +# Memory Update — auth-broker-phase2 — 2026-03-01 + +## Key findings +- Extending broker-based auth to `vault unlock` and `api start` can be done without breaking interactive defaults by using `prompt` as the default broker. +- `api.start_server` can support non-interactive startup unlock by accepting an optional `unlock_password` parameter while preserving existing callers. + +## Patterns / reusable knowledge +- For backward compatibility in CLI tests, keep existing command behavior unchanged unless explicit broker flags are provided. +- Use consistent broker options across commands: `--auth-broker`, `--password-env`, `--broker-service`, `--broker-account`, `--broker-command`. + +## Warnings / gotchas +- When introducing additional args to helper functions used in tests (`start_server`), keep positional compatibility so existing monkeypatched fakes still work. + +# Memory Update — capabilities-command-v1 — 2026-03-01 + +## Key findings +- Agents benefit from a single deterministic capability map command instead of scraping many `--help` screens. +- A stable JSON schema for capabilities is useful for automation and CI checks; text output can remain user-friendly. + +## Patterns / reusable knowledge +- Implement capability output with fixed ordering and versioned schema (`schema_version`) to reduce churn in downstream parsers. +- Expose both CLI and API discovery hints in one surface so agents can bootstrap without external docs. + +## Warnings / gotchas +- Capability maps can drift from actual commands if manually maintained; future work should generate metadata from command registries. + +# Memory Update — bootstrap-context-v1 — 2026-03-01 + +## Key findings +- A dedicated `agent bootstrap-context` command significantly reduces setup overhead for autonomous agents by returning policy/token/auth posture in one deterministic JSON document. +- Treating invalid policy as non-fatal context (`policy.status=invalid`) is useful for diagnostics and recovery workflows. + +## Patterns / reusable knowledge +- Bootstrap context should avoid all sensitive data and include only posture metadata and actionable command hints. +- `sort_keys=True` for JSON output improves reproducibility for automated parsers and snapshot tests. + +## Warnings / gotchas +- Token posture summaries should classify revoked/expired/exhausted separately to avoid misleading “active” counts. + +# Memory Update — posture-check-v1 — 2026-03-01 + +## Key findings +- A lightweight `agent posture-check` command can provide immediate security signal with deterministic output and CI-friendly exit behavior. +- Severity thresholding (`--fail-on`) is useful for phased rollout: start with `critical`, then tighten to `high`/`medium` as posture matures. + +## Patterns / reusable knowledge +- Keep posture findings stable with explicit IDs and severities so automation can suppress or track specific controls over time. +- Treat invalid policy as a critical posture issue but still return full JSON payload for diagnostics. + +## Warnings / gotchas +- Posture checks that read token store should classify revoked/expired/exhausted tokens separately to avoid false positives on active credentials. + +# Memory Update — export-policy-gating-v1 — 2026-03-01 + +## Key findings +- Agent-profile export controls are easiest to roll out by introducing explicit context flags (`--agent-profile` / `X-SeedPass-Agent-Profile`) instead of changing default export behavior for all users. +- Full-vault export and subset export need separate controls; subset export can be deterministic by filtering encrypted index data by policy-allowed kinds before encryption. + +## Patterns / reusable knowledge +- Centralize export control logic in a shared module (`seedpass/core/agent_export_policy.py`) so CLI and API enforcement remain consistent. +- For deterministic subset export payloads, use `json.dumps(..., sort_keys=True, separators=(",",":"))` before encryption. + +## Warnings / gotchas +- Default policy currently allows `totp` in `allow_kinds`; denial tests must provide an explicit restrictive policy fixture when validating blocked TOTP export. diff --git a/src/seedpass/api.py b/src/seedpass/api.py index 43a8db3e..c8922918 100644 --- a/src/seedpass/api.py +++ b/src/seedpass/api.py @@ -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") diff --git a/src/seedpass/cli/__init__.py b/src/seedpass/cli/__init__.py index 989c133c..0eb37512 100644 --- a/src/seedpass/cli/__init__.py +++ b/src/seedpass/cli/__init__.py @@ -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: diff --git a/src/seedpass/cli/agent.py b/src/seedpass/cli/agent.py index f1655020..024be675 100644 --- a/src/seedpass/cli/agent.py +++ b/src/seedpass/cli/agent.py @@ -1,10 +1,15 @@ from __future__ import annotations +import hashlib +import hmac import json import os +import re +import secrets +import socket from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Optional +from typing import Any, Optional import bcrypt import click @@ -14,6 +19,7 @@ from mnemonic import Mnemonic from constants import APP_DIR, initialize_app from seedpass.core.api import EntryService +from seedpass.core.auth_broker import AuthBrokerError, resolve_password as resolve_broker_password from seedpass.core.entry_types import ALL_ENTRY_TYPES, EntryType from seedpass.core.encryption import EncryptionManager from seedpass.core.manager import PasswordManager @@ -32,7 +38,49 @@ PRIVATE_KINDS = { EntryType.NOSTR.value, EntryType.MANAGED_ACCOUNT.value, } + DEFAULT_POLICY = { + "version": 1, + "default_effect": "deny", + "rules": [ + { + "id": "allow_standard_read", + "effect": "allow", + "operations": ["read"], + "kinds": [ + EntryType.PASSWORD.value, + EntryType.TOTP.value, + EntryType.KEY_VALUE.value, + ], + "label_regex": ".*", + "path_regex": "^entry/.*$", + "fields": ["secret", "label", "kind"], + }, + { + "id": "deny_private_material", + "effect": "deny", + "operations": ["read"], + "kinds": sorted(PRIVATE_KINDS), + "label_regex": ".*", + "path_regex": "^entry/.*$", + "fields": ["secret"], + }, + ], + "approvals": { + "require_for": [ + "export", + "reveal_parent_seed", + "private_key_retrieval", + ] + }, + "output": { + "safe_output_default": True, + "redact_fields": ["secret", "value", "private_key", "seed_phrase"], + }, + "export": { + "allow_full_vault": False, + }, + # Legacy compatibility fields kept for older consumers/tests. "allow_kinds": [ EntryType.PASSWORD.value, EntryType.TOTP.value, @@ -42,20 +90,229 @@ DEFAULT_POLICY = { "allow_export_import": False, } +TOKEN_STORE_VERSION = 1 + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + def _policy_path() -> Path: return APP_DIR / "agent_policy.json" -def _deny_all_policy() -> dict: - return { +def _token_store_path() -> Path: + return APP_DIR / "agent_tokens.json" + + +def _audit_key_path() -> Path: + return APP_DIR / "agent_audit.key" + + +def _audit_log_path() -> Path: + return APP_DIR / "agent_audit.log" + + +def _ensure_dir(path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + + +def _mask_secret(value: str | None) -> str | None: + if value is None: + return None + if len(value) <= 4: + return "*" * len(value) + return f"{value[:2]}{'*' * (len(value) - 4)}{value[-2:]}" + + +def _deny_all_policy() -> dict[str, Any]: + deny = { + "version": 1, + "default_effect": "deny", + "rules": [], + "approvals": {"require_for": ["export", "reveal_parent_seed"]}, + "output": { + "safe_output_default": True, + "redact_fields": ["secret", "value", "private_key", "seed_phrase"], + }, + "export": {"allow_full_vault": False}, "allow_kinds": [], "deny_private_reveal": list(ALL_ENTRY_TYPES), "allow_export_import": False, } + return deny -def _load_policy(*, strict: bool = False) -> dict: +def _normalize_legacy_policy(data: dict[str, Any]) -> dict[str, Any]: + """Convert legacy policy keys into the versioned rule model.""" + allow_kinds_raw = data.get("allow_kinds") + deny_private_raw = data.get("deny_private_reveal") + allow_export_import = bool(data.get("allow_export_import", False)) + + allow_kinds = ( + [str(v) for v in allow_kinds_raw if str(v) in ALL_ENTRY_TYPES] + if isinstance(allow_kinds_raw, list) + else list(DEFAULT_POLICY["allow_kinds"]) + ) + deny_private = ( + [str(v) for v in deny_private_raw if str(v) in ALL_ENTRY_TYPES] + if isinstance(deny_private_raw, list) + else list(DEFAULT_POLICY["deny_private_reveal"]) + ) + + policy = dict(DEFAULT_POLICY) + rules: list[dict[str, Any]] = [ + { + "id": "allow_standard_read", + "effect": "allow", + "operations": ["read"], + "kinds": allow_kinds, + "label_regex": ".*", + "path_regex": "^entry/.*$", + "fields": ["secret", "label", "kind"], + } + ] + if deny_private: + rules.append( + { + "id": "deny_private_material", + "effect": "deny", + "operations": ["read"], + "kinds": deny_private, + "label_regex": ".*", + "path_regex": "^entry/.*$", + "fields": ["secret"], + } + ) + policy["rules"] = rules + policy["allow_kinds"] = allow_kinds + policy["deny_private_reveal"] = deny_private + policy["allow_export_import"] = allow_export_import + policy["export"] = {"allow_full_vault": allow_export_import} + policy["output"] = dict(DEFAULT_POLICY["output"]) + return policy + + +def _validate_rule(rule: dict[str, Any]) -> None: + effect = str(rule.get("effect", "")).lower() + if effect not in {"allow", "deny"}: + raise ValueError("rule effect must be 'allow' or 'deny'") + + operations = rule.get("operations", []) + if not isinstance(operations, list) or not operations: + raise ValueError("rule operations must be a non-empty list") + for op in operations: + if not isinstance(op, str): + raise ValueError("rule operations entries must be strings") + + kinds = rule.get("kinds", []) + if not isinstance(kinds, list): + raise ValueError("rule kinds must be a list") + invalid_kinds = [k for k in kinds if str(k) not in ALL_ENTRY_TYPES] + if invalid_kinds: + raise ValueError(f"rule contains invalid kinds: {invalid_kinds}") + + for regex_key in ("label_regex", "path_regex"): + pattern = rule.get(regex_key, ".*") + if not isinstance(pattern, str): + raise ValueError(f"rule {regex_key} must be a string") + try: + re.compile(pattern) + except re.error as exc: # pragma: no cover - exercised via CLI path + raise ValueError(f"invalid {regex_key}: {exc}") from exc + + fields = rule.get("fields", []) + if not isinstance(fields, list) or any(not isinstance(f, str) for f in fields): + raise ValueError("rule fields must be a list of strings") + + +def _normalize_policy(data: dict[str, Any], *, strict: bool) -> dict[str, Any]: + if "rules" not in data: + return _normalize_legacy_policy(data) + + policy = dict(DEFAULT_POLICY) + policy.update({k: v for k, v in data.items() if k in policy}) + + version = int(policy.get("version", 1)) + if version != 1: + raise ValueError("agent policy version must be 1") + + default_effect = str(policy.get("default_effect", "deny")).lower() + if default_effect not in {"allow", "deny"}: + raise ValueError("default_effect must be 'allow' or 'deny'") + policy["default_effect"] = default_effect + + raw_rules = policy.get("rules", []) + if not isinstance(raw_rules, list): + raise ValueError("policy rules must be a list") + + normalized_rules: list[dict[str, Any]] = [] + for rule in raw_rules: + if not isinstance(rule, dict): + raise ValueError("each policy rule must be an object") + _validate_rule(rule) + normalized_rules.append( + { + "id": str(rule.get("id", "rule")), + "effect": str(rule.get("effect", "")).lower(), + "operations": [str(v) for v in rule.get("operations", [])], + "kinds": [str(v) for v in rule.get("kinds", [])], + "label_regex": str(rule.get("label_regex", ".*")), + "path_regex": str(rule.get("path_regex", ".*")), + "fields": [str(v) for v in rule.get("fields", [])], + } + ) + policy["rules"] = normalized_rules + + output_cfg = policy.get("output") + if not isinstance(output_cfg, dict): + output_cfg = {} + output_norm = dict(DEFAULT_POLICY["output"]) + if "safe_output_default" in output_cfg: + output_norm["safe_output_default"] = bool(output_cfg["safe_output_default"]) + redact_fields = output_cfg.get("redact_fields") + if isinstance(redact_fields, list): + output_norm["redact_fields"] = [str(v) for v in redact_fields] + policy["output"] = output_norm + + export_cfg = policy.get("export") + if not isinstance(export_cfg, dict): + export_cfg = {} + policy["export"] = { + "allow_full_vault": bool(export_cfg.get("allow_full_vault", False)) + } + + # Keep compatibility fields in sync with rule model. + allow_kinds = sorted( + { + k + for r in normalized_rules + if r["effect"] == "allow" and "read" in r["operations"] + for k in r["kinds"] + } + ) + deny_private = sorted( + { + k + for r in normalized_rules + if r["effect"] == "deny" and "read" in r["operations"] + for k in r["kinds"] + if k in PRIVATE_KINDS + } + ) + policy["allow_kinds"] = allow_kinds + policy["deny_private_reveal"] = deny_private + policy["allow_export_import"] = bool(policy["export"]["allow_full_vault"]) + + if strict: + # Force full validation for future schema updates. + for rule in policy["rules"]: + _validate_rule(rule) + + return policy + + +def _load_policy(*, strict: bool = False) -> dict[str, Any]: path = _policy_path() if not path.exists(): return dict(DEFAULT_POLICY) @@ -71,42 +328,40 @@ def _load_policy(*, strict: bool = False) -> dict: raise ValueError("agent policy file must be a JSON object") return _deny_all_policy() - policy = dict(DEFAULT_POLICY) - if isinstance(data.get("allow_kinds"), list): - allow_kinds = [str(v) for v in data["allow_kinds"]] - invalid = [v for v in allow_kinds if v not in ALL_ENTRY_TYPES] - if invalid and strict: - raise ValueError(f"agent policy has invalid allow_kinds entries: {invalid}") - policy["allow_kinds"] = [v for v in allow_kinds if v in ALL_ENTRY_TYPES] - if isinstance(data.get("deny_private_reveal"), list): - deny_private = [str(v) for v in data["deny_private_reveal"]] - invalid = [v for v in deny_private if v not in ALL_ENTRY_TYPES] - if invalid and strict: - raise ValueError( - f"agent policy has invalid deny_private_reveal entries: {invalid}" - ) - policy["deny_private_reveal"] = [ - v for v in deny_private if v in ALL_ENTRY_TYPES - ] - if isinstance(data.get("allow_export_import"), bool): - policy["allow_export_import"] = data["allow_export_import"] - return policy + try: + return _normalize_policy(data, strict=strict) + except ValueError: + if strict: + raise + return _deny_all_policy() -def _save_policy(policy: dict) -> None: +def _save_policy(policy: dict[str, Any]) -> None: path = _policy_path() - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(json.dumps(policy, indent=2), encoding="utf-8") + _ensure_dir(path) + normalized = _normalize_policy(policy, strict=True) + path.write_text(json.dumps(normalized, indent=2), encoding="utf-8") os.chmod(path, 0o600) -def _password_from_env(var_name: str) -> str: - value = os.getenv(var_name) - if not value: - raise typer.BadParameter( - f"Missing password env var '{var_name}'. Export it before running this command." +def _agent_password( + *, + broker: str, + password_env: str, + broker_service: str, + broker_account: str, + broker_command: str | None, +) -> str: + try: + return resolve_broker_password( + broker=broker, + password_env=password_env, + broker_service=broker_service, + broker_account=broker_account, + broker_command=broker_command, ) - return value + except AuthBrokerError as exc: + raise typer.BadParameter(str(exc)) from exc def _seed_from_inputs( @@ -133,6 +388,350 @@ def _seed_from_inputs( return seed, bool(generate_seed) +def _load_token_store() -> dict[str, Any]: + path = _token_store_path() + if not path.exists(): + return {"version": TOKEN_STORE_VERSION, "tokens": []} + try: + data = json.loads(path.read_text(encoding="utf-8")) + except Exception: + return {"version": TOKEN_STORE_VERSION, "tokens": []} + if not isinstance(data, dict) or not isinstance(data.get("tokens"), list): + return {"version": TOKEN_STORE_VERSION, "tokens": []} + return data + + +def _save_token_store(store: dict[str, Any]) -> None: + path = _token_store_path() + _ensure_dir(path) + path.write_text(json.dumps(store, indent=2), encoding="utf-8") + os.chmod(path, 0o600) + + +def _hash_token(raw_token: str) -> str: + return hashlib.sha256(raw_token.encode("utf-8")).hexdigest() + + +def _issue_token_record( + name: str, + ttl: int, + scopes: list[str], + kinds: list[str], + label_regex: str, + uses: int, +) -> tuple[str, dict[str, Any]]: + raw = secrets.token_urlsafe(32) + now = _utcnow() + token_id = hashlib.blake2s(raw.encode("utf-8"), digest_size=8).hexdigest() + record = { + "id": token_id, + "name": name, + "token_hash": _hash_token(raw), + "created_at_utc": now.isoformat(), + "expires_at_utc": (now + timedelta(seconds=ttl)).isoformat(), + "revoked_at_utc": None, + "scopes": scopes, + "kinds": kinds, + "label_regex": label_regex, + "uses_remaining": int(uses), + } + return raw, record + + +def _parse_iso8601(value: str) -> datetime: + dt = datetime.fromisoformat(value) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + + +def _validate_token( + raw_token: str, + *, + operation: str, + kind: str, + label: str, + consume_use: bool, +) -> tuple[bool, str, dict[str, Any] | None]: + store = _load_token_store() + digest = _hash_token(raw_token) + for rec in store.get("tokens", []): + if rec.get("token_hash") != digest: + continue + if rec.get("revoked_at_utc"): + return False, "token_revoked", None + exp = rec.get("expires_at_utc") + if not exp or _utcnow() >= _parse_iso8601(exp): + return False, "token_expired", None + + scopes = [str(s) for s in rec.get("scopes", [])] + if operation not in scopes: + return False, "token_scope_denied", None + + kinds = [str(k) for k in rec.get("kinds", [])] + if kinds and kind not in kinds: + return False, "token_kind_denied", None + + label_regex = str(rec.get("label_regex", ".*")) + try: + if not re.search(label_regex, label): + return False, "token_label_denied", None + except re.error: + return False, "token_invalid_regex", None + + uses_remaining = int(rec.get("uses_remaining", 0)) + if uses_remaining <= 0: + return False, "token_exhausted", None + + if consume_use: + rec["uses_remaining"] = uses_remaining - 1 + _save_token_store(store) + + return True, "ok", rec + + return False, "token_not_found", None + + +def _load_audit_key() -> bytes: + path = _audit_key_path() + if path.exists(): + return path.read_bytes() + _ensure_dir(path) + key = secrets.token_bytes(32) + path.write_bytes(key) + os.chmod(path, 0o600) + return key + + +def _append_audit_event(event: str, details: dict[str, Any]) -> None: + key = _load_audit_key() + path = _audit_log_path() + _ensure_dir(path) + + prev_sig = "0" * 64 + if path.exists(): + try: + lines = [ln for ln in path.read_text(encoding="utf-8").splitlines() if ln] + if lines: + prev = json.loads(lines[-1]) + prev_sig = str(prev.get("sig", prev_sig)) + except Exception: + prev_sig = "0" * 64 + + payload = { + "timestamp_utc": _utcnow().isoformat(), + "host": socket.gethostname(), + "event": event, + "details": details, + } + canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")) + sig = hmac.new(key, f"{prev_sig}{canonical}".encode("utf-8"), hashlib.sha256) + payload["sig"] = sig.hexdigest() + + with path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(payload, sort_keys=True) + "\n") + os.chmod(path, 0o600) + + +def _policy_decision( + policy: dict[str, Any], + *, + operation: str, + kind: str, + label: str, + path: str, + field: str, +) -> tuple[bool, str]: + matched_allow = None + for rule in policy.get("rules", []): + operations = set(rule.get("operations", [])) + kinds = set(rule.get("kinds", [])) + fields = set(rule.get("fields", [])) + if operation not in operations: + continue + if kinds and kind not in kinds: + continue + if fields and field not in fields: + continue + try: + if not re.search(str(rule.get("label_regex", ".*")), label): + continue + if not re.search(str(rule.get("path_regex", ".*")), path): + continue + except re.error: + continue + + effect = str(rule.get("effect", "")).lower() + rid = str(rule.get("id", "rule")) + if effect == "deny": + return False, f"policy_deny:{rid}" + if effect == "allow": + matched_allow = rid + + if matched_allow: + return True, f"policy_allow:{matched_allow}" + + if policy.get("default_effect", "deny") == "allow": + return True, "policy_default_allow" + return False, "policy_default_deny" + + +def _build_entry_path(index: int, kind: str, label: str) -> str: + return f"entry/{index}/{kind}/{label}" + + +def _token_posture_summary() -> dict[str, int]: + store = _load_token_store() + now = _utcnow() + total = 0 + active = 0 + revoked = 0 + expired = 0 + exhausted = 0 + for rec in store.get("tokens", []): + total += 1 + if rec.get("revoked_at_utc"): + revoked += 1 + continue + exp = rec.get("expires_at_utc") + if exp: + try: + if _parse_iso8601(exp) <= now: + expired += 1 + continue + except Exception: + expired += 1 + continue + if int(rec.get("uses_remaining", 0)) <= 0: + exhausted += 1 + continue + active += 1 + return { + "total": total, + "active": active, + "revoked": revoked, + "expired": expired, + "exhausted": exhausted, + } + + +def _severity_rank(severity: str) -> int: + order = {"info": 0, "low": 1, "medium": 2, "high": 3, "critical": 4} + return order.get(severity, 0) + + +def _posture_findings(policy: dict[str, Any], policy_status: str) -> list[dict[str, Any]]: + findings: list[dict[str, Any]] = [] + if policy_status != "ok": + findings.append( + { + "id": "policy_invalid", + "severity": "critical", + "message": "Agent policy JSON is invalid; default deny fallback may block operations.", + "remediation": "Run `seedpass agent policy-lint` and fix policy syntax/schema.", + } + ) + + if str(policy.get("default_effect", "deny")).lower() == "allow": + findings.append( + { + "id": "policy_default_allow", + "severity": "high", + "message": "Policy default effect is allow; unmatched operations are permitted.", + "remediation": "Set `default_effect` to `deny` and add explicit allow rules.", + } + ) + + if len(policy.get("rules", [])) == 0: + findings.append( + { + "id": "policy_no_rules", + "severity": "medium", + "message": "Policy has no rules; behavior depends entirely on default effect.", + "remediation": "Define explicit allow/deny rules for read/create/update/export.", + } + ) + + if bool(policy.get("allow_export_import", False)): + findings.append( + { + "id": "export_import_allowed", + "severity": "high", + "message": "Agent export/import is enabled.", + "remediation": "Disable export/import for agent profiles unless explicitly required.", + } + ) + + if not bool(policy.get("output", {}).get("safe_output_default", True)): + findings.append( + { + "id": "safe_output_disabled", + "severity": "high", + "message": "Safe output default is disabled; secrets may leak into logs/stdout.", + "remediation": "Set output.safe_output_default to true.", + } + ) + + store = _load_token_store() + now = _utcnow() + long_lived = 0 + broad_scope = 0 + for rec in store.get("tokens", []): + if rec.get("revoked_at_utc"): + continue + exp = rec.get("expires_at_utc") + created = rec.get("created_at_utc") + try: + if exp and created: + ttl = (_parse_iso8601(exp) - _parse_iso8601(created)).total_seconds() + if ttl > 86400: + long_lived += 1 + except Exception: + pass + + scopes = [str(v) for v in rec.get("scopes", [])] + if "export" in scopes or "reveal_parent_seed" in scopes: + broad_scope += 1 + elif not scopes: + broad_scope += 1 + + if exp: + try: + if _parse_iso8601(exp) <= now: + findings.append( + { + "id": "expired_tokens_present", + "severity": "low", + "message": "Expired tokens are present in token store.", + "remediation": "Periodically prune expired token records.", + } + ) + break + except Exception: + pass + + if long_lived: + findings.append( + { + "id": "long_lived_tokens", + "severity": "medium", + "message": f"{long_lived} active token(s) exceed 24h TTL.", + "remediation": "Reduce token TTLs and rotate often.", + } + ) + if broad_scope: + findings.append( + { + "id": "overbroad_token_scopes", + "severity": "high", + "message": f"{broad_scope} token(s) include broad or risky scopes.", + "remediation": "Issue least-privilege tokens (prefer read-only, narrow kinds/labels).", + } + ) + + return findings + + @app.command("init") def agent_init( seed: Optional[str] = typer.Option( @@ -147,6 +746,27 @@ def agent_init( password_env: str = typer.Option( "SEEDPASS_PASSWORD", "--password-env", help="Env var containing master password" ), + auth_broker: str = typer.Option( + "env", + "--auth-broker", + help="Non-interactive password broker (env|keyring|command)", + click_type=click.Choice(["env", "keyring", "command"], case_sensitive=False), + ), + broker_service: str = typer.Option( + "seedpass", + "--broker-service", + help="Keyring service name when --auth-broker=keyring", + ), + broker_account: str = typer.Option( + "default", + "--broker-account", + help="Keyring account (or logical broker identity)", + ), + broker_command: Optional[str] = typer.Option( + None, + "--broker-command", + help="Command that prints password to stdout for --auth-broker=command", + ), switch_existing: bool = typer.Option( False, "--switch-existing", @@ -161,7 +781,13 @@ def agent_init( ) -> None: """Initialize a profile non-interactively for agent workflows.""" initialize_app() - password = _password_from_env(password_env) + password = _agent_password( + broker=auth_broker, + password_env=password_env, + broker_service=broker_service, + broker_account=broker_account, + broker_command=broker_command, + ) seed_phrase, generated = _seed_from_inputs(seed, seed_file, generate_seed) fp_mgr = FingerprintManager(APP_DIR) @@ -222,6 +848,13 @@ def agent_policy_show() -> None: typer.echo(json.dumps(policy, indent=2)) +@app.command("policy-lint") +def agent_policy_lint() -> None: + """Validate the policy file and print a normalized representation.""" + policy = _load_policy(strict=True) + typer.echo(json.dumps({"status": "ok", "policy": policy}, indent=2)) + + @app.command("policy-set") def agent_policy_set( allow_kind: list[str] = typer.Option( @@ -241,18 +874,258 @@ def agent_policy_set( "--allow-export-import", help="Allow export/import operations in agent mode", ), + safe_output_default: bool = typer.Option( + True, + "--safe-output-default/--unsafe-output-default", + help="Mask sensitive fields in output by default", + ), ) -> None: - """Set policy flags used by agent commands.""" + """Set policy flags used by agent commands (legacy-compatible helper).""" policy = _load_policy(strict=False) if allow_kind: policy["allow_kinds"] = list(allow_kind) if deny_private_kind: policy["deny_private_reveal"] = list(deny_private_kind) policy["allow_export_import"] = bool(allow_export_import) + + policy = _normalize_legacy_policy(policy) + policy["output"]["safe_output_default"] = bool(safe_output_default) _save_policy(policy) typer.echo(json.dumps(policy, indent=2)) +@app.command("token-issue") +def agent_token_issue( + name: str = typer.Option("agent", "--name", help="Human-readable token name"), + scope: list[str] = typer.Option( + ["read"], + "--scope", + help="Allowed operation scope (repeatable, e.g. read/export)", + ), + kind: list[str] = typer.Option( + None, + "--kind", + help="Restrict token to entry kinds (repeatable)", + click_type=click.Choice(ALL_ENTRY_TYPES), + ), + label_regex: str = typer.Option( + ".*", + "--label-regex", + help="Regex constraint for entry labels", + ), + ttl: int = typer.Option( + 300, + "--ttl", + min=1, + help="Token lifetime in seconds", + ), + uses: int = typer.Option( + 1, + "--uses", + min=1, + help="Maximum number of successful uses", + ), +) -> None: + """Issue a short-lived scoped token for agent operations.""" + try: + re.compile(label_regex) + except re.error as exc: + raise typer.BadParameter(f"Invalid --label-regex: {exc}") from exc + + scopes = [str(s).strip().lower() for s in scope if str(s).strip()] + if not scopes: + raise typer.BadParameter("At least one --scope is required") + + kinds = list(kind or []) + raw, record = _issue_token_record(name, ttl, scopes, kinds, label_regex, uses) + + store = _load_token_store() + store.setdefault("version", TOKEN_STORE_VERSION) + store.setdefault("tokens", []) + store["tokens"].append(record) + _save_token_store(store) + + _append_audit_event( + "agent_token_issued", + { + "id": record["id"], + "name": name, + "scopes": scopes, + "kinds": kinds, + "expires_at_utc": record["expires_at_utc"], + "uses": uses, + }, + ) + + typer.echo( + json.dumps( + { + "status": "ok", + "token": raw, + "token_id": record["id"], + "name": name, + "scopes": scopes, + "kinds": kinds, + "label_regex": label_regex, + "expires_at_utc": record["expires_at_utc"], + "uses_remaining": uses, + }, + indent=2, + ) + ) + + +@app.command("token-list") +def agent_token_list(show_revoked: bool = typer.Option(False, "--show-revoked")) -> None: + """List issued tokens without exposing raw token material.""" + store = _load_token_store() + tokens = [] + for rec in store.get("tokens", []): + if not show_revoked and rec.get("revoked_at_utc"): + continue + tokens.append( + { + "id": rec.get("id"), + "name": rec.get("name"), + "created_at_utc": rec.get("created_at_utc"), + "expires_at_utc": rec.get("expires_at_utc"), + "revoked_at_utc": rec.get("revoked_at_utc"), + "scopes": rec.get("scopes", []), + "kinds": rec.get("kinds", []), + "label_regex": rec.get("label_regex", ".*"), + "uses_remaining": rec.get("uses_remaining", 0), + } + ) + typer.echo(json.dumps({"status": "ok", "tokens": tokens}, indent=2)) + + +@app.command("token-revoke") +def agent_token_revoke(token_id: str = typer.Argument(..., help="Token id to revoke")) -> None: + """Revoke an issued token by id.""" + store = _load_token_store() + now = _utcnow().isoformat() + changed = False + for rec in store.get("tokens", []): + if rec.get("id") == token_id and not rec.get("revoked_at_utc"): + rec["revoked_at_utc"] = now + changed = True + break + + if not changed: + typer.echo(json.dumps({"status": "error", "reason": "token_not_found"}, indent=2)) + raise typer.Exit(1) + + _save_token_store(store) + _append_audit_event("agent_token_revoked", {"id": token_id}) + typer.echo(json.dumps({"status": "ok", "token_id": token_id}, indent=2)) + + +@app.command("bootstrap-context") +def agent_bootstrap_context(ctx: typer.Context) -> None: + """Return deterministic context for autonomous agent bootstrap.""" + fingerprint = (ctx.obj or {}).get("fingerprint") + try: + policy = _load_policy(strict=True) + policy = _normalize_policy(policy, strict=False) + policy_status = "ok" + policy_error = None + except ValueError as exc: + policy = _deny_all_policy() + policy_status = "invalid" + policy_error = str(exc) + + payload = { + "status": "ok", + "context_schema_version": 1, + "fingerprint": fingerprint, + "policy": { + "status": policy_status, + "error": policy_error, + "version": int(policy.get("version", 1)), + "default_effect": str(policy.get("default_effect", "deny")), + "rule_count": len(policy.get("rules", [])), + "safe_output_default": bool( + policy.get("output", {}).get("safe_output_default", True) + ), + "allow_export_import": bool(policy.get("allow_export_import", False)), + }, + "tokens": _token_posture_summary(), + "auth_brokers": { + "supported": ["env", "keyring", "command"], + "default_agent_get": "env", + "default_agent_init": "env", + }, + "commands": { + "policy": ["agent policy-show", "agent policy-lint", "agent policy-set"], + "tokens": ["agent token-issue", "agent token-list", "agent token-revoke"], + "secret_access": ["agent get --fingerprint "], + "discovery": ["seedpass capabilities --format json", "seedpass --help"], + }, + } + typer.echo(json.dumps(payload, indent=2, sort_keys=True)) + + +@app.command("posture-check") +def agent_posture_check( + fail_on: str = typer.Option( + "critical", + "--fail-on", + help="Minimum severity that returns non-zero exit code", + click_type=click.Choice( + ["info", "low", "medium", "high", "critical"], case_sensitive=False + ), + ), + format: str = typer.Option( + "json", + "--format", + help="Output format", + click_type=click.Choice(["json", "text"], case_sensitive=False), + ), +) -> None: + """Run security posture checks for agent configuration and token hygiene.""" + try: + policy = _load_policy(strict=True) + policy = _normalize_policy(policy, strict=False) + policy_status = "ok" + except ValueError: + policy = _deny_all_policy() + policy_status = "invalid" + + findings = _posture_findings(policy, policy_status) + highest = "info" + for finding in findings: + if _severity_rank(finding["severity"]) > _severity_rank(highest): + highest = finding["severity"] + + payload = { + "status": "ok", + "check": "agent_posture", + "policy_status": policy_status, + "token_summary": _token_posture_summary(), + "finding_count": len(findings), + "highest_severity": highest, + "fail_on": fail_on.lower(), + "findings": findings, + } + + if format.lower() == "json": + typer.echo(json.dumps(payload, indent=2, sort_keys=True)) + else: + typer.echo("SeedPass Agent Posture Check") + typer.echo(f"policy_status: {payload['policy_status']}") + typer.echo(f"highest_severity: {payload['highest_severity']}") + typer.echo(f"finding_count: {payload['finding_count']}") + if findings: + typer.echo("") + for finding in findings: + typer.echo( + f"- [{finding['severity']}] {finding['id']}: {finding['message']}" + ) + + if _severity_rank(highest) >= _severity_rank(fail_on.lower()): + raise typer.Exit(code=1) + + @app.command("get") def agent_get( ctx: typer.Context, @@ -260,31 +1133,67 @@ def agent_get( password_env: str = typer.Option( "SEEDPASS_PASSWORD", "--password-env", help="Env var containing master password" ), + auth_broker: str = typer.Option( + "env", + "--auth-broker", + help="Non-interactive password broker (env|keyring|command)", + click_type=click.Choice(["env", "keyring", "command"], case_sensitive=False), + ), + 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 fingerprint)", + ), + broker_command: Optional[str] = typer.Option( + None, + "--broker-command", + help="Command that prints password to stdout for --auth-broker=command", + ), + token: Optional[str] = typer.Option( + None, + "--token", + help="Scoped agent token. Defaults to SEEDPASS_AGENT_TOKEN env var.", + ), ttl: int = typer.Option( 30, "--ttl", min=1, help="Lease time in seconds for the returned secret metadata", ), + reveal: bool = typer.Option( + False, + "--reveal", + help="Return plaintext secret instead of redacted output", + ), ) -> None: - """Retrieve one secret as JSON with policy enforcement.""" + """Retrieve one secret as JSON with policy enforcement, redaction, and auditing.""" fingerprint = (ctx.obj or {}).get("fingerprint") if not fingerprint: raise typer.BadParameter("Specify target profile with --fingerprint.") - password = _password_from_env(password_env) + password = _agent_password( + broker=auth_broker, + password_env=password_env, + broker_service=broker_service, + broker_account=broker_account or str(fingerprint), + broker_command=broker_command, + ) + raw_token = token or os.getenv("SEEDPASS_AGENT_TOKEN") + try: policy = _load_policy(strict=True) + policy = _normalize_policy(policy, strict=False) except ValueError as exc: - typer.echo( - json.dumps( - { - "status": "denied", - "reason": "invalid_policy", - "detail": str(exc), - }, - indent=2, - ) - ) + payload = { + "status": "denied", + "reason": "invalid_policy", + "detail": str(exc), + } + typer.echo(json.dumps(payload, indent=2)) raise typer.Exit(1) pm = PasswordManager(fingerprint=fingerprint, password=password) @@ -307,43 +1216,84 @@ def agent_get( for idx, label, username, url, archived, etype in matches ], } + _append_audit_event( + "agent_secret_access_denied", + { + "fingerprint": fingerprint, + "query": query, + "reason": "ambiguous_or_missing", + "match_count": len(matches), + }, + ) typer.echo(json.dumps(payload, indent=2)) raise typer.Exit(1) index = matches[0][0] entry = service.retrieve_entry(index) - kind = entry.get("type", entry.get("kind")) - if kind is None: + kind = str(entry.get("type", entry.get("kind", ""))) + label = str(entry.get("label", "")) + if not kind: raise typer.BadParameter("Entry kind missing.") - allow_kinds = set(policy.get("allow_kinds", [])) - deny_private = set(policy.get("deny_private_reveal", [])) - if allow_kinds and kind not in allow_kinds: - typer.echo( - json.dumps( - { - "status": "denied", - "reason": "kind_not_allowed", - "kind": kind, - "policy": policy, - }, - indent=2, - ) + entry_path = _build_entry_path(index, kind, label) + allowed, decision = _policy_decision( + policy, + operation="read", + kind=kind, + label=label, + path=entry_path, + field="secret", + ) + if not allowed: + payload = { + "status": "denied", + "reason": decision, + "kind": kind, + "label": label, + "index": index, + } + _append_audit_event( + "agent_secret_access_denied", + { + "fingerprint": fingerprint, + "index": index, + "kind": kind, + "label": label, + "reason": decision, + }, ) + typer.echo(json.dumps(payload, indent=2)) raise typer.Exit(1) - if kind in deny_private: - typer.echo( - json.dumps( - { - "status": "denied", - "reason": "private_kind_blocked", - "kind": kind, - "policy": policy, - }, - indent=2, - ) + + token_meta: dict[str, Any] | None = None + if raw_token: + ok, token_reason, token_meta = _validate_token( + raw_token, + operation="read", + kind=kind, + label=label, + consume_use=True, ) - raise typer.Exit(1) + if not ok: + payload = { + "status": "denied", + "reason": token_reason, + "kind": kind, + "label": label, + "index": index, + } + _append_audit_event( + "agent_secret_access_denied", + { + "fingerprint": fingerprint, + "index": index, + "kind": kind, + "label": label, + "reason": token_reason, + }, + ) + typer.echo(json.dumps(payload, indent=2)) + raise typer.Exit(1) secret: Optional[str] = None if kind == EntryType.PASSWORD.value: @@ -351,30 +1301,48 @@ def agent_get( elif kind == EntryType.TOTP.value: secret = service.get_totp_code(index) elif kind == EntryType.KEY_VALUE.value: - secret = entry.get("value") + secret = str(entry.get("value", "")) else: - typer.echo( - json.dumps( - { - "status": "error", - "reason": "unsupported_kind_for_agent_get", - "kind": kind, - }, - indent=2, - ) - ) + payload = { + "status": "error", + "reason": "unsupported_kind_for_agent_get", + "kind": kind, + } + typer.echo(json.dumps(payload, indent=2)) raise typer.Exit(1) - now = datetime.now(timezone.utc) + now = _utcnow() + safe_default = bool(policy.get("output", {}).get("safe_output_default", True)) + redacted = safe_default and not reveal + output_secret = _mask_secret(secret) if redacted else secret + payload = { "status": "ok", "fingerprint": fingerprint, "index": index, "kind": kind, - "label": entry.get("label"), + "label": label, "lease_ttl_sec": ttl, "issued_at_utc": now.isoformat(), "expires_at_utc": (now + timedelta(seconds=ttl)).isoformat(), - "secret": secret, + "safe_output": redacted, + "secret": output_secret, + "policy_decision": decision, } + if token_meta is not None: + payload["token_id"] = token_meta.get("id") + payload["token_uses_remaining"] = token_meta.get("uses_remaining", 0) + + _append_audit_event( + "agent_secret_access_granted", + { + "fingerprint": fingerprint, + "index": index, + "kind": kind, + "label": label, + "safe_output": redacted, + "token_id": (token_meta or {}).get("id"), + }, + ) + typer.echo(json.dumps(payload, indent=2)) diff --git a/src/seedpass/cli/api.py b/src/seedpass/cli/api.py index 618d8d9a..1365a38a 100644 --- a/src/seedpass/cli/api.py +++ b/src/seedpass/cli/api.py @@ -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." ) diff --git a/src/seedpass/cli/capabilities.py b/src/seedpass/cli/capabilities.py new file mode 100644 index 00000000..b8dc1a21 --- /dev/null +++ b/src/seedpass/cli/capabilities.py @@ -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 "], + "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 --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())) diff --git a/src/seedpass/cli/entry.py b/src/seedpass/cli/entry.py index 1474cdfb..ff1d11b4 100644 --- a/src/seedpass/cli/entry.py +++ b/src/seedpass/cli/entry.py @@ -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)) diff --git a/src/seedpass/cli/vault.py b/src/seedpass/cli/vault.py index a5269340..38a87773 100644 --- a/src/seedpass/cli/vault.py +++ b/src/seedpass/cli/vault.py @@ -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 --help`) to discover available features." + ) @app.command("lock") diff --git a/src/seedpass/core/agent_export_policy.py b/src/seedpass/core/agent_export_policy.py new file mode 100644 index 00000000..97976814 --- /dev/null +++ b/src/seedpass/core/agent_export_policy.py @@ -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, + } diff --git a/src/seedpass/core/auth_broker.py b/src/seedpass/core/auth_broker.py new file mode 100644 index 00000000..69e1f786 --- /dev/null +++ b/src/seedpass/core/auth_broker.py @@ -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}'.") diff --git a/src/tests/test_api_new_endpoints.py b/src/tests/test_api_new_endpoints.py index daf87fc2..42f8f673 100644 --- a/src/tests/test_api_new_endpoints.py +++ b/src/tests/test_api_new_endpoints.py @@ -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 diff --git a/src/tests/test_cli_agent_mode.py b/src/tests/test_cli_agent_mode.py index 294a0f64..9f72a73c 100644 --- a/src/tests/test_cli_agent_mode.py +++ b/src/tests/test_cli_agent_mode.py @@ -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 diff --git a/src/tests/test_cli_core_services.py b/src/tests/test_cli_core_services.py index d81178ce..24e4e83c 100644 --- a/src/tests/test_cli_core_services.py +++ b/src/tests/test_cli_core_services.py @@ -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 = {} diff --git a/src/tests/test_typer_cli.py b/src/tests/test_typer_cli.py index 6e60b9b9..6bd20521 100644 --- a/src/tests/test_typer_cli.py +++ b/src/tests/test_typer_cli.py @@ -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 = {} diff --git a/torch/.scheduler-memory/latest/daily/store.ok b/torch/.scheduler-memory/latest/daily/store.ok index 1c3d28c8..1e40bb67 100644 --- a/torch/.scheduler-memory/latest/daily/store.ok +++ b/torch/.scheduler-memory/latest/daily/store.ok @@ -1 +1 @@ -dummy memory +MEMORY_STORED diff --git a/torch/task-logs/daily/2026-02-28T22-22-56Z__todo-triage-agent__completed.md b/torch/task-logs/daily/2026-02-28T22-22-56Z__todo-triage-agent__completed.md new file mode 100644 index 00000000..192e2334 --- /dev/null +++ b/torch/task-logs/daily/2026-02-28T22-22-56Z__todo-triage-agent__completed.md @@ -0,0 +1,23 @@ +--- +agent: todo-triage-agent +cadence: daily +status: completed +timestamp: 2026-02-28T22:22:56Z +platform: codex +--- + +# Run Summary + +The `todo-triage-agent` successfully scanned the codebase for TODO/FIXME/XXX markers and generated an inventory artifact. + +## Findings +- `artifacts/todos.txt`: Inventory of TODOs found in the codebase. + +## Actions Taken +- A significant TODO was identified in `src/utils/memory_protection.py`: `# TODO: Replace this Python implementation with a Rust/WASM module for critical cryptographic operations.` +- This TODO represents a large, security-sensitive task (P0/P2) that requires a dedicated issue or project, falling outside the scope of a trivial fix. +- Created `artifacts/todo-triage-issues.txt` with the issue details. +- Left the high-risk TODO for human triage/issue creation as per protocol. + +## Next Steps +- Review `artifacts/todo-triage-issues.txt` and create a GitHub issue for the memory protection Rust/WASM implementation. diff --git a/torch/task-logs/daily/2026-03-01T00-10-34Z__scheduler-update-agent__completed.md b/torch/task-logs/daily/2026-03-01T00-10-34Z__scheduler-update-agent__completed.md new file mode 100644 index 00000000..b562401e --- /dev/null +++ b/torch/task-logs/daily/2026-03-01T00-10-34Z__scheduler-update-agent__completed.md @@ -0,0 +1,6 @@ +--- +agent: scheduler-update-agent +platform: codex +--- +Status: Success +Learnings: MEMORY_STORED diff --git a/torch/task-logs/daily/temp-memory.md b/torch/task-logs/daily/temp-memory.md deleted file mode 100644 index 548d358a..00000000 --- a/torch/task-logs/daily/temp-memory.md +++ /dev/null @@ -1 +0,0 @@ -All oversized files had recent refactor commits; skipped decomposition.