Merge branch 'main' into beta

# Conflicts:
#	memory-update.md
This commit is contained in:
thePR0M3TH3AN committed 2026-03-01 15:44:22 -05:00
commit 982fba6404
41 files changed
+2895 -144

No files matched your search

+58 -9
View File
@@ -70,6 +70,7 @@ SeedPass now uses the `portalocker` library for cross-platform file locking. No
- **Quick Unlock:** Optionally skip the password prompt after verifying once. - **Quick Unlock:** Optionally skip the password prompt after verifying once.
- **Secret Mode:** When enabled, newly generated and retrieved passwords are copied to your clipboard and automatically cleared after a delay. - **Secret Mode:** When enabled, newly generated and retrieved passwords are copied to your clipboard and automatically cleared after a delay.
- **Tagging Support:** Organize entries with optional tags and find them quickly via search. - **Tagging Support:** Organize entries with optional tags and find them quickly via search.
- **Entry Graph Links:** Connect entries with typed relationships (`target_id`, `relation`, `note`) for knowledge-base style navigation.
- **Typed Search Results:** Results now display each entry's type for quicker identification. - **Typed Search Results:** Results now display each entry's type for quicker identification.
- **Manual Vault Export/Import:** Create encrypted backups or restore them using the CLI or API. - **Manual Vault Export/Import:** Create encrypted backups or restore them using the CLI or API.
- **Parent Seed Backup:** Securely save an encrypted copy of the master seed. - **Parent Seed Backup:** Securely save an encrypted copy of the master seed.
@@ -98,6 +99,7 @@ with policy and security guardrails:
- **Safer automation primitives:** `agent job-run`, `agent job-template`, and job profiles with policy-stamp checks and signed template manifests. - **Safer automation primitives:** `agent job-run`, `agent job-template`, and job profiles with policy-stamp checks and signed template manifests.
- **Recovery hardening:** `agent recovery-split/recover/drill/drill-list` for Shamir split/recover and signed drill evidence. - **Recovery hardening:** `agent recovery-split/recover/drill/drill-list` for Shamir split/recover and signed drill evidence.
- **Deterministic export controls:** policy-filtered exports plus manifest integrity checks with `agent export-check` and `agent export-manifest-verify`. - **Deterministic export controls:** policy-filtered exports plus manifest integrity checks with `agent export-check` and `agent export-manifest-verify`.
- **Agent document I/O:** policy-gated filesystem import/export with `agent document-import` and `agent document-export`.
- **Security posture tooling:** `agent posture-check` and `agent posture-remediate` for drift detection and remediation bundles. - **Security posture tooling:** `agent posture-check` and `agent posture-remediate` for drift detection and remediation bundles.
- **Auditable access chain:** append-only HMAC-chained audit records with verification support. - **Auditable access chain:** append-only HMAC-chained audit records with verification support.
@@ -501,6 +503,13 @@ export SEEDPASS_PASSWORD='your-strong-password'
seedpass -f <fingerprint> agent get github --ttl 30 seedpass -f <fingerprint> agent get github --ttl 30
``` ```
4. Import/export document entries from automation (requires policy `allow_export_import=true`):
```bash
seedpass -f <fingerprint> agent document-import --file ./notes.md --label "Ops Notes"
seedpass -f <fingerprint> agent document-export --entry-id 42 --out ./exports
```
By default, agent policy blocks private key/seed-style entry kinds from being By default, agent policy blocks private key/seed-style entry kinds from being
revealed through `agent get`. revealed through `agent get`.
@@ -609,19 +618,59 @@ SeedPass supports storing more than just passwords and 2FA secrets. You can also
- **Nostr Key Pair** – store the index used to derive an `npub`/`nsec` pair for Nostr clients. When you retrieve one of these entries, SeedPass can display QR codes for the keys. The `npub` is wrapped in the `nostr:` URI scheme so any client can scan it, while the `nsec` QR is shown only after a security warning. - **Nostr Key Pair** – store the index used to derive an `npub`/`nsec` pair for Nostr clients. When you retrieve one of these entries, SeedPass can display QR codes for the keys. The `npub` is wrapped in the `nostr:` URI scheme so any client can scan it, while the `nsec` QR is shown only after a security warning.
- **Key/Value** – store a simple key and value for miscellaneous secrets or configuration data. - **Key/Value** – store a simple key and value for miscellaneous secrets or configuration data.
- **Managed Account** – derive a child seed under the current profile. Loading a managed account switches to a nested profile and the header shows `<parent_fp> > Managed Account > <child_fp>`. Press Enter on the main menu to return to the parent profile. - **Managed Account** – derive a child seed under the current profile. Loading a managed account switches to a nested profile and the header shows `<parent_fp> > Managed Account > <child_fp>`. Press Enter on the main menu to return to the parent profile.
- **Document** – store editable text content plus a file type/extension (`md`, `txt`, `py`, `js`, `csv`, etc.) with a built-in terminal editor (`p`, `a`, `i <n>`, `e <n>`, `d <n>`, `w`, `q`).
The table below summarizes the extra fields stored for each entry type. Every entry includes a `label`, while only password entries track a `url`. All entry types include:
- `type` and `kind`
- `label` (title)
- `archived`
- `date_added` (UTC ISO-8601)
- `date_modified` (UTC ISO-8601)
- `links` (optional graph edges: `target_id`, `relation`, `note`)
The table below summarizes additional kind-specific fields.
| Entry Type | Extra Fields | | Entry Type | Extra Fields |
|-----------------|-------------------------------------------------------------------------------------------------------------------------------------------------------| |-----------------|-------------------------------------------------------------------------------------------------------------------------------------------------------|
| Password | `username`, `url`, `length`, `archived`, optional `notes`, optional `custom_fields` (may include hidden fields), optional `tags` | | Password | `username`, `url`, `length`, optional `notes`, optional `custom_fields` (may include hidden fields), optional `tags` |
| 2FA (TOTP) | `index` or `secret`, `period`, `digits`, `archived`, optional `notes`, optional `tags` | | 2FA (TOTP) | `index` or `secret`, `period`, `digits`, optional `notes`, optional `tags` |
| SSH Key | `index`, `archived`, optional `notes`, optional `tags` | | SSH Key | `index`, optional `notes`, optional `tags` |
| Seed Phrase | `index`, `word_count` *(mnemonic regenerated; never stored)*, `archived`, optional `notes`, optional `tags` | | Seed Phrase | `index`, `word_count` *(mnemonic regenerated; never stored)*, optional `notes`, optional `tags` |
| PGP Key | `index`, `key_type`, `archived`, optional `user_id`, optional `notes`, optional `tags` | | PGP Key | `index`, `key_type`, optional `user_id`, optional `notes`, optional `tags` |
| Nostr Key Pair | `index`, `archived`, optional `notes`, optional `tags` | | Nostr Key Pair | `index`, optional `notes`, optional `tags` |
| Key/Value | `key`, `value`, `archived`, optional `notes`, optional `custom_fields`, optional `tags` | | Key/Value | `key`, `value`, optional `notes`, optional `custom_fields`, optional `tags` |
| Managed Account | `index`, `word_count`, `fingerprint`, `archived`, optional `notes`, optional `tags` | | Managed Account | `index`, `word_count`, `fingerprint`, optional `notes`, optional `tags` |
| Document | `content`, `file_type`, optional `notes`, optional `custom_fields`, optional `tags` |
For a complete per-kind schema reference, see [docs/entry_types.md](docs/entry_types.md).
### Document Import/Export
SeedPass can import local text files as `document` entries and export stored documents back to your OS filesystem.
- CLI import:
- `seedpass entry import-document --file ./notes.md --label "Ops Notes"`
- CLI export:
- `seedpass entry export-document --entry-id 42 --out ./exports`
- API import:
- `POST /api/v1/entry/document/import` with JSON: `{"path":"./notes.md","label":"Ops Notes"}`
- API export:
- `POST /api/v1/entry/{entry_id}/document/export` with JSON: `{"path":"./exports","overwrite":false}`
In TUI, choose `Add Entry > Document > Import Document File` to import, and use `Entry Actions > Export Document to File` on a document entry to export.
### Entry Graph Links
SeedPass entries can be connected with typed links so tags can evolve into an explicit knowledge graph.
- CLI:
- `seedpass entry link-add --entry-id 1 --target-id 2 --relation references --note "used by runbook"`
- `seedpass entry links --entry-id 1`
- `seedpass entry link-remove --entry-id 1 --target-id 2 --relation references`
- API:
- `GET /api/v1/entry/{id}/links`
- `POST /api/v1/entry/{id}/links`
- `DELETE /api/v1/entry/{id}/links`
### Managing Multiple Seeds ### Managing Multiple Seeds
+2
View File
@@ -21,6 +21,8 @@ This directory contains SeedPass project documentation.
- `ARCHITECTURE.md` - `ARCHITECTURE.md`
- `SPEC.md` - `SPEC.md`
- `entry_types.md`
- `entry_graph.md`
- `security.md` - `security.md`
- `nostr_setup.md` - `nostr_setup.md`
- `nostr_namespace_reset.md` - `nostr_namespace_reset.md`
+53
View File
@@ -0,0 +1,53 @@
# Entry Graph Links
SeedPass now supports explicit relationships between entries, so your vault can evolve from tag clusters into a typed knowledge graph.
## Link Model
Every entry can include:
- `links`: list of objects with:
- `target_id`: target entry index
- `relation`: link type (for example `references`, `depends_on`, `owned_by`, `derived_from`)
- `note`: optional free-form context
Example:
```json
{
"links": [
{ "target_id": 42, "relation": "references", "note": "Used by deploy script" }
]
}
```
## CLI Commands
- Add link:
- `seedpass entry link-add --entry-id 1 --target-id 42 --relation references --note "Used by deploy script"`
- List links:
- `seedpass entry links --entry-id 1`
- Remove link:
- `seedpass entry link-remove --entry-id 1 --target-id 42 --relation references`
- Replace full link list:
- `seedpass entry modify 1 --links-json '[{"target_id":42,"relation":"references","note":"Used by deploy script"}]'`
## API Endpoints
- `GET /api/v1/entry/{id}/links`
- `POST /api/v1/entry/{id}/links`
- Body: `{ "target_id": 42, "relation": "references", "note": "..." }`
- `DELETE /api/v1/entry/{id}/links`
- Body: `{ "target_id": 42, "relation": "references" }`
- `PUT /api/v1/entry/{id}` also accepts a full `links` array for replacement.
## Search and Sync Behavior
- Search now matches link metadata (`relation`, `note`) and linked target labels.
- Sync conflict resolution includes `links` in deterministic equal-timestamp union merge, alongside `tags` and `custom_fields`.
## Recommended Conventions
- Keep `relation` values short and consistent (`references`, `depends_on`, `owned_by`, `belongs_to`, `derived_from`).
- Use `note` for human context instead of encoding meaning into long relation strings.
- Use tags for broad grouping, links for explicit semantics.
+110
View File
@@ -0,0 +1,110 @@
# Entry Types and Fields
This document defines the entry kinds supported by SeedPass, shared fields, and kind-specific fields used by the TUI and API.
For graph-link workflows and relationship modeling, see `docs/entry_graph.md`.
## Shared Fields
Every entry kind stores:
- `type`: canonical kind identifier.
- `kind`: kind alias used for backward compatibility.
- `label`: entry title.
- `archived`: boolean archive flag.
- `date_added`: UTC ISO-8601 timestamp set on creation.
- `date_modified`: UTC ISO-8601 timestamp updated on modification.
- `notes`: optional free-form notes.
- `tags`: optional string list.
- `links`: optional relationship list (`target_id`, `relation`, `note`) for graph-style connections between entries.
Internally, SeedPass also tracks `modified_ts` (UNIX timestamp) for deterministic merge conflict resolution.
## Kind-Specific Fields
### `password`
- `length`
- `username`
- `url`
- `policy` (optional password policy overrides)
- `custom_fields` (optional hidden/plain additional fields)
### `totp`
- `index` (for deterministic TOTP) or `secret` (for imported TOTP)
- `period`
- `digits`
- `deterministic`
### `ssh`
- `index` (derivation index)
### `seed`
- `index` (derivation index)
- `word_count`
### `pgp`
- `index` (derivation index)
- `key_type`
- `user_id`
### `nostr`
- `index` (derivation index)
### `key_value`
- `key`
- `value`
- `custom_fields` (optional)
### `managed_account`
- `index` (derivation index)
- `word_count`
- `fingerprint`
- `custom_fields` (optional)
### `document`
- `content`
- `file_type` (extension/type like `txt`, `md`, `py`, `js`, `csv`)
- `custom_fields` (optional)
## TUI Notes
- Add flow: `Main Menu > Add Entry > Document`.
- Document editing uses a built-in line editor:
- `p` print content
- `a` append line
- `i <n>` insert before line
- `e <n>` edit line
- `d <n>` delete line
- `w` save
- `q` cancel
## API Notes
- Create: `POST /api/v1/entry` with `type=document` (or `kind=document`)
- Update: `PUT /api/v1/entry/{id}` with `content`, `file_type`, `label`, `notes`, `tags`, `archived`
- Read: `GET /api/v1/entry/{id}` returns all stored fields, including `date_added` and `date_modified`
- File import: `POST /api/v1/entry/document/import` with `path` (+ optional `label`, `notes`, `tags`, `archived`)
- File export: `POST /api/v1/entry/{id}/document/export` with optional `path` and `overwrite`
- Link list: `GET /api/v1/entry/{id}/links`
- Link add: `POST /api/v1/entry/{id}/links` with `{ "target_id": 2, "relation": "references", "note": "..." }`
- Link remove: `DELETE /api/v1/entry/{id}/links` with `{ "target_id": 2, "relation": "references" }`
## CLI Notes
- Add directly: `seedpass entry add-document --label "Doc" --content "text" --file-type txt`
- Import from file: `seedpass entry import-document --file ./doc.md`
- Export to file: `seedpass entry export-document --entry-id 42 --out ./exports`
- Agent import: `seedpass -f <fp> agent document-import --file ./doc.md --token <token>`
- Agent export: `seedpass -f <fp> agent document-export --entry-id 42 --out ./exports --token <token>`
- Link add: `seedpass entry link-add --entry-id 1 --target-id 2 --relation references --note "supports deployment"`
- Link list: `seedpass entry links --entry-id 1`
- Link remove: `seedpass entry link-remove --entry-id 1 --target-id 2 --relation references`
+2 -2
View File
@@ -22,8 +22,8 @@ Per-entry merge rules:
1. Higher `modified_ts` wins. 1. Higher `modified_ts` wins.
2. If `modified_ts` is equal, tie-break by canonical entry hash. 2. If `modified_ts` is equal, tie-break by canonical entry hash.
3. On equal timestamp, selected fields are union-merged deterministically: 3. On equal timestamp, selected fields are union-merged deterministically:
- `tags` and `custom_fields` use canonical sorted union. - `tags`, `custom_fields`, and `links` use canonical sorted union.
- Optional kind-specific fields are backfilled when one side is empty. - Optional kind-specific fields (for example document `content`/`file_type`) are backfilled when one side is empty.
4. `archived` at equal timestamp uses conservative OR semantics. 4. `archived` at equal timestamp uses conservative OR semantics.
## Tombstone Semantics ## Tombstone Semantics
+2
View File
@@ -54,6 +54,8 @@
<li><a href="#" data-file="README.md" class="doc-link active">&gt; README</a></li> <li><a href="#" data-file="README.md" class="doc-link active">&gt; README</a></li>
<li><a href="#" data-file="ARCHITECTURE.md" class="doc-link">&gt; Architecture</a></li> <li><a href="#" data-file="ARCHITECTURE.md" class="doc-link">&gt; Architecture</a></li>
<li><a href="#" data-file="SPEC.md" class="doc-link">&gt; Specification</a></li> <li><a href="#" data-file="SPEC.md" class="doc-link">&gt; Specification</a></li>
<li><a href="#" data-file="entry_types.md" class="doc-link">&gt; Entry Types</a></li>
<li><a href="#" data-file="entry_graph.md" class="doc-link">&gt; Entry Graph</a></li>
<li><a href="#" data-file="security.md" class="doc-link">&gt; Security</a></li> <li><a href="#" data-file="security.md" class="doc-link">&gt; Security</a></li>
<li><a href="#" data-file="security_readiness_checklist.md" class="doc-link">&gt; Security Readiness</a></li> <li><a href="#" data-file="security_readiness_checklist.md" class="doc-link">&gt; Security Readiness</a></li>
<li><a href="#" data-file="agent_autonomy_security_plan.md" class="doc-link">&gt; Agent Security Plan</a></li> <li><a href="#" data-file="agent_autonomy_security_plan.md" class="doc-link">&gt; Agent Security Plan</a></li>
+38
View File
@@ -66,6 +66,7 @@
<div>├─► <i class="fas fa-network-wired"></i> SSH Keys</div> <div>├─► <i class="fas fa-network-wired"></i> SSH Keys</div>
<div>├─► <i class="fas fa-lock"></i> PGP Key</div> <div>├─► <i class="fas fa-lock"></i> PGP Key</div>
<div>├─► <i class="fas fa-bolt"></i> Nostr Keys</div> <div>├─► <i class="fas fa-bolt"></i> Nostr Keys</div>
<div>├─► <i class="fas fa-file-lines"></i> Documents</div>
<div>└─► <i class="fas fa-users-cog"></i> Managed Accounts</div> <div>└─► <i class="fas fa-users-cog"></i> Managed Accounts</div>
</div> </div>
</div> </div>
@@ -122,6 +123,21 @@
<h3>Secret Mode</h3> <h3>Secret Mode</h3>
<p>Copy to clipboard directly, clearing automatically after delay.</p> <p>Copy to clipboard directly, clearing automatically after delay.</p>
</div> </div>
<div class="feature-item">
<i class="fas fa-file-code"></i>
<h3>Document Entries</h3>
<p>Store rich text content with file types like <code>md</code>, <code>txt</code>, <code>py</code>, and edit in-terminal.</p>
</div>
<div class="feature-item">
<i class="fas fa-clock-rotate-left"></i>
<h3>Entry Timestamps</h3>
<p>All entry kinds include <code>date_added</code> and <code>date_modified</code> for deterministic auditability.</p>
</div>
<div class="feature-item">
<i class="fas fa-diagram-project"></i>
<h3>Entry Graph Links</h3>
<p>Create typed relationships between entries using <code>target_id</code>, <code>relation</code>, and <code>note</code>.</p>
</div>
</div> </div>
</section> </section>
@@ -169,6 +185,16 @@
<h3>Audit, Posture, Recovery</h3> <h3>Audit, Posture, Recovery</h3>
<p>Chained audit logs, posture checks/remediation, Shamir split-drills, and deterministic export validation.</p> <p>Chained audit logs, posture checks/remediation, Shamir split-drills, and deterministic export validation.</p>
</div> </div>
<div class="feature-item">
<i class="fas fa-file-import"></i>
<h3>Agent Document I/O</h3>
<p>Policy-gated <code>agent document-import</code> and <code>agent document-export</code> for controlled filesystem workflows.</p>
</div>
<div class="feature-item">
<i class="fas fa-share-nodes"></i>
<h3>Graph-Aware Entry Ops</h3>
<p>Automate knowledge links with <code>entry link-add</code>, <code>entry links</code>, and <code>entry link-remove</code>.</p>
</div>
</div> </div>
</section> </section>
@@ -191,6 +217,18 @@
</div> </div>
</div> </div>
<div class="info-blocks mt-2">
<div class="info-block">
<h3><span class="accent">&gt;</span> Knowledge Graph Example</h3>
<pre class="graph-snippet">[12] Ops Runbook (document)
├─ references ──> [7] API Token (key_value)
├─ depends_on ──> [3] Deploy SSH (ssh)
└─ rotates_with -> [9] Prod MFA (totp)
Query: "depends_on" or "deploy ssh" -> returns entry 12</pre>
</div>
</div>
<div class="info-blocks"> <div class="info-blocks">
<div class="info-block"> <div class="info-block">
<h3><span class="accent">&gt;</span> Recovery via Nostr</h3> <h3><span class="accent">&gt;</span> Recovery via Nostr</h3>
+13
View File
@@ -567,6 +567,19 @@ a:hover {
color: var(--text-muted); color: var(--text-muted);
} }
.graph-snippet {
margin-top: 10px;
padding: 12px;
border: 1px solid var(--border-color);
border-left: 2px solid var(--accent-secondary);
border-radius: 4px;
background: rgba(8, 10, 12, 0.45);
color: var(--text-primary);
font-size: 0.82rem;
line-height: 1.55;
overflow-x: auto;
}
/* Footer */ /* Footer */
.tui-footer { .tui-footer {
display: flex; display: flex;
+125
View File
@@ -37,3 +37,128 @@
- Verified with local venv: - Verified with local venv:
- `./.venv/bin/pytest -q src/tests/test_post_sync_messages.py` - `./.venv/bin/pytest -q src/tests/test_post_sync_messages.py`
- `./.venv/bin/pytest -q src/tests/test_background_vault_sync_paths.py` - `./.venv/bin/pytest -q src/tests/test_background_vault_sync_paths.py`
## Entry model expansion: document kind + universal date fields
- Added a new `document` entry kind across core enum, TUI, API, CLI, and agent secret resolution.
- Added universal `date_added` and `date_modified` fields for all entry kinds.
- Creation now sets both fields.
- Modification preserves `date_added` and updates `date_modified`.
- Legacy entries are normalized in `_load_index()` by deriving date fields from `modified_ts`/`updated`.
- TUI now supports `Document` creation via a built-in line editor with basic commands (`p`, `a`, `i <n>`, `e <n>`, `d <n>`, `w`, `q`).
- Added document-specific display and edit flows in manager/display services.
## Sync/policy compatibility notes
- Updated deterministic merge kind matrix to include document fields (`content`, `file_type`) for equal-timestamp backfill behavior.
- Agent defaults now include `document` in non-private allow-kinds, and agent secret resolution returns document content for `document` entries.
## Tests and docs
- Added/updated tests for:
- document entry roundtrip and CLI add command
- universal date field presence/update semantics across all entry kinds
- API create/modify behavior for `document`
- large document Nostr sync roundtrip
- Added `docs/entry_types.md` and updated README/docs index and sync conflict docs.
- Verified with:
- `./.venv/bin/pytest -q src/tests/test_cli_entry_add_commands.py src/tests/test_entry_add.py src/tests/test_entry_types_roundtrip.py src/tests/test_modify_entry_validation.py src/tests/test_api.py src/tests/test_api_new_endpoints.py src/tests/test_full_sync_roundtrip.py`
- Result: `141 passed`.
## Document file import/export workflows
- Added document file I/O at core layer:
- `EntryManager.import_document_file(path, ...)`
- `EntryManager.export_document_file(index, output_path=None, overwrite=False)`
- Added TUI support:
- `Add Entry > Document` now supports `New Document` and `Import Document File`.
- Document entry actions now include `Export Document to File`.
- Added automation surfaces:
- CLI: `entry import-document`, `entry export-document`
- API: `POST /api/v1/entry/document/import`, `POST /api/v1/entry/{id}/document/export`
- Regression coverage added for core file roundtrip, CLI commands, and API endpoints.
## Agent document import/export commands
- Added dedicated agent CLI commands:
- `seedpass agent document-import --file <path>`
- `seedpass agent document-export <query>` (or `--entry-id`)
- Enforcement model for both commands:
- Requires `--fingerprint` and non-interactive auth broker password resolution.
- Denies unless policy `allow_export_import` is enabled.
- Denies unless `document` kind is policy-allowed (`allow_kinds` / kind export check).
- Supports scoped token enforcement with operation scopes (`import` for import, `export` for export).
- Export honors approval gates when policy requires `export` approval.
- Emits structured JSON responses and append-only audit events for granted/denied outcomes.
- Discovery updates:
- Added `document_io` command hints to `agent bootstrap-context` payload.
- Added `document_io` section to `seedpass capabilities --format json` security features.
- Added CLI tests for:
- token-scoped agent document import success path
- token-scoped agent document export success path
- bootstrap context includes document command hints
- Environment note: could not run pytest in this shell because `pytest` is unavailable (`python3 -m pytest` reports module missing).
## Test environment note (2026-03-01)
- Repo `.venv` already had `pytest` and `typer`; upgraded to `pytest 9.0.2` and `typer 0.24.1`.
- New agent document I/O tests initially failed due mocked policy normalization (empty `rules` caused normalized `allow_kinds` to become empty).
- Fixed tests by adding explicit allow rule for `document` in mocked policies.
- Verification run:
- `.venv/bin/pytest -q src/tests/test_document_file_io.py src/tests/test_cli_entry_add_commands.py src/tests/test_api_new_endpoints.py src/tests/test_entry_types_roundtrip.py src/tests/test_full_sync_roundtrip.py src/tests/test_cli_agent_mode.py`
- Result: `150 passed`.
## Full-suite regression repair after document rollout
- Ran full suite after enabling agent document I/O and found 12 regressions (mostly stale tests).
- Fixed regressions by updating tests for:
- new document add-menu option/mocks (`handle_add_document`),
- expanded `EntryService.modify_entry` kwargs,
- universal `date_added`/`date_modified` and `custom_fields` fields in entry expectations,
- `list_entries(sort_by="updated")` relying on `modified_ts`,
- legacy decrypt fallback monkeypatches to accept `salt` argument and revised iteration-call expectations.
- Verification: `.venv/bin/pytest -q src/tests` => `891 passed, 11 skipped`.
## Knowledge graph links integration (entry relationships)
- Added first-class graph edges to every entry via shared `links` field (list of `{target_id, relation, note}`).
- Implemented core link operations in `EntryManager`:
- `add_link(entry_id, target_id, relation, note)`
- `remove_link(entry_id, target_id, relation=None)`
- `get_links(entry_id)` with resolved target label/kind.
- Added normalization/defaulting for `links` in index load path, including cache-return normalization to prevent stale in-memory schemas.
- Extended `modify_entry(..., links=...)` to allow full links replacement with validation.
- Search now matches link metadata (`relation`, `note`) and linked target labels, enabling graph-style discovery from existing index scans.
- Sync conflict equal-timestamp union now includes `links` alongside `tags` and `custom_fields` for deterministic convergence.
- Added API endpoints:
- `GET /api/v1/entry/{id}/links`
- `POST /api/v1/entry/{id}/links`
- `DELETE /api/v1/entry/{id}/links`
and `PUT /api/v1/entry/{id}` now accepts `links`.
- Added CLI graph commands:
- `seedpass entry link-add`
- `seedpass entry links`
- `seedpass entry link-remove`
plus `entry modify --links-json`.
- Documentation updated in `README.md`, `docs/entry_types.md`, and `docs/sync_conflict_contract.md`.
- Validation:
- Focused suite: `140 passed`
- Full suite: `895 passed, 11 skipped`.
## Website/docs refresh for graph features
- Added dedicated graph docs page: `docs/entry_graph.md`.
- Updated docs indexes/navigation:
- `docs/README.md` now lists `entry_graph.md`.
- `landing/docs.html` sidebar now includes `Entry Graph`.
- Updated landing page feature cards (`landing/index.html`) to highlight:
- Entry Graph Links
- Graph-Aware Entry Ops CLI commands
- Updated docs cross-links:
- `docs/entry_types.md` points to `docs/entry_graph.md`.
- Updated README feature bullets to include Entry Graph Links.
## Landing graph example + help-hint refresh
- Added a visual graph snippet block to landing architecture section (`landing/index.html`) showing linked entries and graph query behavior.
- Styled graph snippet in `landing/style.css` (`.graph-snippet`) to match the terminal UI aesthetic.
- Expanded help/discovery surfaces for users and agents:
- `seedpass capabilities` now includes a `knowledge_graph` security feature block with CLI/API graph commands and sync/search behavior.
- Capabilities API discovery now lists `/api/v1/entry/{id}/links`.
- Capabilities text output now explicitly mentions graph links and document workflows.
- Capabilities `help_hints` now include graph/document command pointers.
- `agent bootstrap-context` now includes `commands.knowledge_graph` hints.
- Root CLI help string now mentions document I/O and entry graph links.
- Verification:
- `.venv/bin/pytest -q src/tests/test_typer_cli.py -k "capabilities_text or capabilities_json" src/tests/test_cli_agent_mode.py -k "bootstrap_context"`
- Result: `2 passed`.
+5 -3
View File
@@ -425,9 +425,7 @@ def handle_post_to_nostr(
logging.info("Encrypted index posted to Nostr successfully.") logging.info("Encrypted index posted to Nostr successfully.")
else: else:
client = getattr(password_manager, "nostr_client", None) client = getattr(password_manager, "nostr_client", None)
detail = ( detail = getattr(client, "last_error", None) if client is not None else None
getattr(client, "last_error", None) if client is not None else None
)
if detail: if detail:
print(colored(f"\N{CROSS MARK} Sync failed… {detail}", "red")) print(colored(f"\N{CROSS MARK} Sync failed… {detail}", "red"))
logging.error("Failed to post encrypted index to Nostr: %s", detail) logging.error("Failed to post encrypted index to Nostr: %s", detail)
@@ -1413,6 +1411,7 @@ def display_menu(
print(color_text("6. PGP Key", "menu")) print(color_text("6. PGP Key", "menu"))
print(color_text("7. Key/Value", "menu")) print(color_text("7. Key/Value", "menu"))
print(color_text("8. Managed Account", "menu")) print(color_text("8. Managed Account", "menu"))
print(color_text("9. Document", "menu"))
sub_choice = input( sub_choice = input(
"Select entry type or press Enter to go back: " "Select entry type or press Enter to go back: "
).strip() ).strip()
@@ -1441,6 +1440,9 @@ def display_menu(
elif sub_choice == "8": elif sub_choice == "8":
password_manager.handle_add_managed_account() password_manager.handle_add_managed_account()
break break
elif sub_choice == "9":
password_manager.handle_add_document()
break
elif not sub_choice: elif not sub_choice:
break break
else: else:
+191 -19
View File
@@ -1021,55 +1021,78 @@ async def create_entry(
int(entry.get("length", 12)), int(entry.get("length", 12)),
entry.get("username"), entry.get("username"),
entry.get("url"), entry.get("url"),
archived=entry.get("archived", False),
notes=entry.get("notes", ""),
custom_fields=entry.get("custom_fields"),
tags=entry.get("tags"),
**kwargs, **kwargs,
) )
return {"id": index} return {"id": index}
if etype == "totp": if etype == "totp":
index = pm.entry_manager.get_next_index() index = pm.entry_manager.get_next_index()
totp_kwargs: dict[str, Any] = {
"secret": entry.get("secret"),
"index": entry.get("index"),
"period": int(entry.get("period", 30)),
"digits": int(entry.get("digits", 6)),
"notes": entry.get("notes", ""),
"archived": entry.get("archived", False),
"deterministic": entry.get("deterministic", False),
}
if entry.get("tags") is not None:
totp_kwargs["tags"] = entry.get("tags")
uri = pm.entry_manager.add_totp( uri = pm.entry_manager.add_totp(
entry.get("label"), entry.get("label"),
pm.KEY_TOTP_DET if entry.get("deterministic", False) else None, pm.KEY_TOTP_DET if entry.get("deterministic", False) else None,
secret=entry.get("secret"), **totp_kwargs,
index=entry.get("index"),
period=int(entry.get("period", 30)),
digits=int(entry.get("digits", 6)),
notes=entry.get("notes", ""),
archived=entry.get("archived", False),
deterministic=entry.get("deterministic", False),
) )
return {"id": index, "uri": uri} return {"id": index, "uri": uri}
if etype == "ssh": if etype == "ssh":
ssh_kwargs: dict[str, Any] = {
"index": entry.get("index"),
"notes": entry.get("notes", ""),
"archived": entry.get("archived", False),
}
if entry.get("tags") is not None:
ssh_kwargs["tags"] = entry.get("tags")
index = pm.entry_manager.add_ssh_key( index = pm.entry_manager.add_ssh_key(
entry.get("label"), entry.get("label"),
pm.parent_seed, pm.parent_seed,
index=entry.get("index"), **ssh_kwargs,
notes=entry.get("notes", ""),
archived=entry.get("archived", False),
) )
return {"id": index} return {"id": index}
if etype == "pgp": if etype == "pgp":
pgp_kwargs: dict[str, Any] = {
"index": entry.get("index"),
"key_type": entry.get("key_type", "ed25519"),
"user_id": entry.get("user_id", ""),
"notes": entry.get("notes", ""),
"archived": entry.get("archived", False),
}
if entry.get("tags") is not None:
pgp_kwargs["tags"] = entry.get("tags")
index = pm.entry_manager.add_pgp_key( index = pm.entry_manager.add_pgp_key(
entry.get("label"), entry.get("label"),
pm.parent_seed, pm.parent_seed,
index=entry.get("index"), **pgp_kwargs,
key_type=entry.get("key_type", "ed25519"),
user_id=entry.get("user_id", ""),
notes=entry.get("notes", ""),
archived=entry.get("archived", False),
) )
return {"id": index} return {"id": index}
if etype == "nostr": if etype == "nostr":
nostr_kwargs: dict[str, Any] = {
"index": entry.get("index"),
"notes": entry.get("notes", ""),
"archived": entry.get("archived", False),
}
if entry.get("tags") is not None:
nostr_kwargs["tags"] = entry.get("tags")
index = pm.entry_manager.add_nostr_key( index = pm.entry_manager.add_nostr_key(
entry.get("label"), entry.get("label"),
pm.parent_seed, pm.parent_seed,
index=entry.get("index"), **nostr_kwargs,
notes=entry.get("notes", ""),
archived=entry.get("archived", False),
) )
return {"id": index} return {"id": index}
@@ -1079,6 +1102,25 @@ async def create_entry(
entry.get("key"), entry.get("key"),
entry.get("value"), entry.get("value"),
notes=entry.get("notes", ""), notes=entry.get("notes", ""),
custom_fields=entry.get("custom_fields"),
tags=entry.get("tags"),
archived=entry.get("archived", False),
)
return {"id": index}
if etype == "document":
label = entry.get("label") or entry.get("title")
document_kwargs: dict[str, Any] = {
"file_type": entry.get("file_type", "txt"),
"notes": entry.get("notes", ""),
"archived": entry.get("archived", False),
}
if entry.get("tags") is not None:
document_kwargs["tags"] = entry.get("tags")
index = pm.entry_manager.add_document(
label,
entry.get("content", ""),
**document_kwargs,
) )
return {"id": index} return {"id": index}
@@ -1093,6 +1135,8 @@ async def create_entry(
pm.parent_seed, pm.parent_seed,
index=entry.get("index"), index=entry.get("index"),
notes=entry.get("notes", ""), notes=entry.get("notes", ""),
archived=entry.get("archived", False),
tags=entry.get("tags"),
) )
return {"id": index} return {"id": index}
@@ -1125,12 +1169,140 @@ def update_entry(
digits=entry.get("digits"), digits=entry.get("digits"),
key=entry.get("key"), key=entry.get("key"),
value=entry.get("value"), value=entry.get("value"),
content=entry.get("content"),
file_type=entry.get("file_type"),
custom_fields=entry.get("custom_fields"),
tags=entry.get("tags"),
links=entry.get("links"),
archived=entry.get("archived"),
) )
except ValueError as e: except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) raise HTTPException(status_code=400, detail=str(e))
return {"status": "ok"} return {"status": "ok"}
@app.get("/api/v1/entry/{entry_id}/links")
def get_entry_links(
request: Request,
entry_id: int,
authorization: str | None = Header(None),
) -> dict[str, Any]:
"""Return resolved graph links for an entry."""
_check_token(request, authorization)
_require_unlocked(request)
pm = _get_pm(request)
try:
links = pm.entry_manager.get_links(entry_id)
except ValueError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
return {"entry_id": entry_id, "links": links}
@app.post("/api/v1/entry/{entry_id}/links")
def add_entry_link(
request: Request,
entry_id: int,
data: dict,
authorization: str | None = Header(None),
) -> dict[str, Any]:
"""Add a graph link from one entry to another."""
_check_token(request, authorization)
_require_unlocked(request)
pm = _get_pm(request)
if "target_id" not in data:
raise HTTPException(status_code=400, detail="missing_target_id")
try:
links = pm.entry_manager.add_link(
entry_id,
int(data.get("target_id")),
relation=str(data.get("relation", "related_to")),
note=str(data.get("note", "")),
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return {"entry_id": entry_id, "links": links}
@app.delete("/api/v1/entry/{entry_id}/links")
def remove_entry_link(
request: Request,
entry_id: int,
data: dict,
authorization: str | None = Header(None),
) -> dict[str, Any]:
"""Remove graph links from one entry to a target entry."""
_check_token(request, authorization)
_require_unlocked(request)
pm = _get_pm(request)
if "target_id" not in data:
raise HTTPException(status_code=400, detail="missing_target_id")
try:
links = pm.entry_manager.remove_link(
entry_id,
int(data.get("target_id")),
relation=data.get("relation"),
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return {"entry_id": entry_id, "links": links}
@app.post("/api/v1/entry/document/import")
def import_document_entry(
request: Request,
data: dict,
authorization: str | None = Header(None),
) -> dict[str, Any]:
"""Import a local text file as a document entry."""
_check_token(request, authorization)
_require_unlocked(request)
pm = _get_pm(request)
source_path = data.get("path")
if not source_path:
raise HTTPException(status_code=400, detail="missing_path")
try:
index = pm.entry_manager.import_document_file(
source_path,
label=data.get("label"),
notes=data.get("notes", ""),
archived=bool(data.get("archived", False)),
tags=data.get("tags"),
)
except FileNotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return {"id": index}
@app.post("/api/v1/entry/{entry_id}/document/export")
def export_document_entry(
request: Request,
entry_id: int,
data: dict,
authorization: str | None = Header(None),
) -> dict[str, Any]:
"""Export a document entry to a local file path."""
_check_token(request, authorization)
_require_unlocked(request)
pm = _get_pm(request)
try:
dest = pm.entry_manager.export_document_file(
entry_id,
output_path=data.get("path"),
overwrite=bool(data.get("overwrite", False)),
)
except FileExistsError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return {"path": str(dest)}
@app.post("/api/v1/entry/{entry_id}/archive") @app.post("/api/v1/entry/{entry_id}/archive")
def archive_entry( def archive_entry(
request: Request, entry_id: int, authorization: str | None = Header(None) request: Request, entry_id: int, authorization: str | None = Header(None)
+2 -1
View File
@@ -16,7 +16,8 @@ app = typer.Typer(
help=( help=(
"SeedPass command line interface. " "SeedPass command line interface. "
"Run `seedpass capabilities` for a deterministic feature map " "Run `seedpass capabilities` for a deterministic feature map "
"and `seedpass <group> --help` for group-specific commands." "and `seedpass <group> --help` for group-specific commands, "
"including document I/O and entry graph links."
), ),
invoke_without_command=True, invoke_without_command=True,
) )
+484 -1
View File
@@ -113,6 +113,7 @@ DEFAULT_POLICY = {
EntryType.PASSWORD.value, EntryType.PASSWORD.value,
EntryType.TOTP.value, EntryType.TOTP.value,
EntryType.KEY_VALUE.value, EntryType.KEY_VALUE.value,
EntryType.DOCUMENT.value,
], ],
"label_regex": ".*", "label_regex": ".*",
"path_regex": "^entry/.*$", "path_regex": "^entry/.*$",
@@ -152,6 +153,7 @@ DEFAULT_POLICY = {
EntryType.PASSWORD.value, EntryType.PASSWORD.value,
EntryType.TOTP.value, EntryType.TOTP.value,
EntryType.KEY_VALUE.value, EntryType.KEY_VALUE.value,
EntryType.DOCUMENT.value,
], ],
"deny_private_reveal": sorted(PRIVATE_KINDS), "deny_private_reveal": sorted(PRIVATE_KINDS),
"allow_export_import": False, "allow_export_import": False,
@@ -267,6 +269,8 @@ def _resolve_secret_for_kind(
return service.get_totp_code(index) return service.get_totp_code(index)
if kind == EntryType.KEY_VALUE.value: if kind == EntryType.KEY_VALUE.value:
return str(entry.get("value", "")) return str(entry.get("value", ""))
if kind == EntryType.DOCUMENT.value:
return str(entry.get("content", ""))
if kind == EntryType.SEED.value: if kind == EntryType.SEED.value:
return pm.entry_manager.get_seed_phrase(index, pm.parent_seed) return pm.entry_manager.get_seed_phrase(index, pm.parent_seed)
if kind == EntryType.MANAGED_ACCOUNT.value: if kind == EntryType.MANAGED_ACCOUNT.value:
@@ -1302,7 +1306,9 @@ def _remediation_actions(findings: list[dict[str, Any]]) -> list[dict[str, Any]]
"export_import_allowed": { "export_import_allowed": {
"title": "Disable broad export/import for agent profiles", "title": "Disable broad export/import for agent profiles",
"commands": ["seedpass agent policy-review --file <candidate.json>"], "commands": ["seedpass agent policy-review --file <candidate.json>"],
"notes": ["Set export.allow_full_vault=false and keep allow_export_import=false."], "notes": [
"Set export.allow_full_vault=false and keep allow_export_import=false."
],
}, },
"secret_isolation_disabled": { "secret_isolation_disabled": {
"title": "Enable high-risk secret isolation", "title": "Enable high-risk secret isolation",
@@ -3290,6 +3296,15 @@ def agent_bootstrap_context(ctx: typer.Context) -> None:
"agent export-check", "agent export-check",
"agent export-manifest-verify", "agent export-manifest-verify",
], ],
"document_io": [
"agent document-import --file <path>",
"agent document-export <query>",
],
"knowledge_graph": [
"entry link-add --entry-id <source> --target-id <target>",
"entry links --entry-id <source>",
"entry link-remove --entry-id <source> --target-id <target>",
],
"posture": ["agent posture-check", "agent posture-remediate"], "posture": ["agent posture-check", "agent posture-remediate"],
"secret_access": ["agent get --fingerprint <fp> <query>"], "secret_access": ["agent get --fingerprint <fp> <query>"],
"discovery": ["seedpass capabilities --format json", "seedpass --help"], "discovery": ["seedpass capabilities --format json", "seedpass --help"],
@@ -3657,6 +3672,474 @@ def agent_export_manifest_verify(
raise typer.Exit(code=1) raise typer.Exit(code=1)
@app.command("document-import")
def agent_document_import(
ctx: typer.Context,
file: Path = typer.Option(..., "--file", help="Input file path"),
label: Optional[str] = typer.Option(None, "--label", help="Document title override"),
notes: str = typer.Option("", "--notes", help="Entry notes"),
tag: list[str] = typer.Option(
[], "--tag", help="Tag to attach (repeatable)", show_default=False
),
archived: bool = typer.Option(False, "--archived", help="Import as archived"),
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.",
),
) -> None:
"""Import a local file as a document entry with agent policy enforcement."""
fingerprint = (ctx.obj or {}).get("fingerprint")
if not fingerprint:
raise typer.BadParameter("Specify target profile with --fingerprint.")
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:
payload = {
"status": "denied",
"reason": "invalid_policy",
"detail": str(exc),
}
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
if not bool(policy.get("allow_export_import", False)):
payload = {
"status": "denied",
"reason": "policy_deny:export_import_disabled",
"kind": EntryType.DOCUMENT.value,
"file": str(file),
}
_append_audit_event(
"agent_document_import_denied",
{
"fingerprint": fingerprint,
"kind": EntryType.DOCUMENT.value,
"file": str(file),
"reason": "policy_deny:export_import_disabled",
},
)
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
allowed, decision = evaluate_kind_export(policy, EntryType.DOCUMENT.value)
if not allowed:
payload = {
"status": "denied",
"reason": decision,
"kind": EntryType.DOCUMENT.value,
"file": str(file),
}
_append_audit_event(
"agent_document_import_denied",
{
"fingerprint": fingerprint,
"kind": EntryType.DOCUMENT.value,
"file": str(file),
"reason": decision,
},
)
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
token_meta: dict[str, Any] | None = None
if raw_token:
ok, token_reason, token_meta = _validate_token(
raw_token,
operation="import",
kind=EntryType.DOCUMENT.value,
label=label or file.stem,
consume_use=True,
)
if not ok:
payload = {
"status": "denied",
"reason": token_reason,
"kind": EntryType.DOCUMENT.value,
"file": str(file),
}
_append_audit_event(
"agent_document_import_denied",
{
"fingerprint": fingerprint,
"kind": EntryType.DOCUMENT.value,
"file": str(file),
"reason": token_reason,
},
)
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
pm = PasswordManager(fingerprint=fingerprint, password=password)
service = EntryService(pm)
try:
index = service.import_document_file(
file,
label=label,
notes=notes,
tags=tag,
archived=archived,
)
entry = service.retrieve_entry(index) or {}
except Exception as exc:
payload = {
"status": "error",
"reason": "document_import_failed",
"detail": str(exc),
"file": str(file),
}
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
payload = {
"status": "ok",
"fingerprint": fingerprint,
"index": index,
"kind": EntryType.DOCUMENT.value,
"label": entry.get("label", label or file.stem),
"file_type": entry.get("file_type", file.suffix.lstrip(".").lower()),
"source_file": str(file),
"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_document_import_granted",
{
"fingerprint": fingerprint,
"index": index,
"kind": EntryType.DOCUMENT.value,
"label": payload["label"],
"source_file": str(file),
"token_id": (token_meta or {}).get("id"),
},
)
typer.echo(json.dumps(payload, indent=2))
@app.command("document-export")
def agent_document_export(
ctx: typer.Context,
query: Optional[str] = typer.Argument(
None, help="Document label or index query (omit when using --entry-id)"
),
entry_id: Optional[int] = typer.Option(
None, "--entry-id", help="Document entry index override"
),
out: Optional[str] = typer.Option(
None,
"--out",
help="Output file path or directory (default: current directory)",
),
overwrite: bool = typer.Option(False, "--overwrite", help="Overwrite if exists"),
approval_id: Optional[str] = typer.Option(
None,
"--approval-id",
help="Approval id when policy requires export approval",
),
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.",
),
) -> None:
"""Export a document entry to a local file with agent policy enforcement."""
fingerprint = (ctx.obj or {}).get("fingerprint")
if not fingerprint:
raise typer.BadParameter("Specify target profile with --fingerprint.")
if entry_id is None and not query:
raise typer.BadParameter("Provide query or --entry-id.")
if entry_id is not None and query:
raise typer.BadParameter("Provide either query or --entry-id, not both.")
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:
payload = {
"status": "denied",
"reason": "invalid_policy",
"detail": str(exc),
}
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
if not bool(policy.get("allow_export_import", False)):
payload = {
"status": "denied",
"reason": "policy_deny:export_import_disabled",
"kind": EntryType.DOCUMENT.value,
}
_append_audit_event(
"agent_document_export_denied",
{
"fingerprint": fingerprint,
"kind": EntryType.DOCUMENT.value,
"reason": "policy_deny:export_import_disabled",
},
)
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
allowed, decision = evaluate_kind_export(policy, EntryType.DOCUMENT.value)
if not allowed:
payload = {
"status": "denied",
"reason": decision,
"kind": EntryType.DOCUMENT.value,
}
_append_audit_event(
"agent_document_export_denied",
{
"fingerprint": fingerprint,
"kind": EntryType.DOCUMENT.value,
"reason": decision,
},
)
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
pm = PasswordManager(fingerprint=fingerprint, password=password)
service = EntryService(pm)
if entry_id is None:
assert query is not None
matches = service.search_entries(query, kinds=[EntryType.DOCUMENT.value])
if len(matches) != 1:
payload = {
"status": "error",
"reason": "ambiguous_or_missing",
"match_count": len(matches),
"matches": [
{
"index": idx,
"label": label_v,
"kind": etype.value,
}
for idx, label_v, _username, _url, _archived, etype in matches
],
}
_append_audit_event(
"agent_document_export_denied",
{
"fingerprint": fingerprint,
"query": query,
"reason": "ambiguous_or_missing",
"match_count": len(matches),
},
)
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
entry_id = int(matches[0][0])
entry = service.retrieve_entry(entry_id)
if not isinstance(entry, dict):
payload = {"status": "error", "reason": "entry_not_found", "index": entry_id}
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
kind = str(entry.get("type", entry.get("kind", "")))
label = str(entry.get("label", ""))
if kind != EntryType.DOCUMENT.value:
payload = {
"status": "denied",
"reason": "entry_kind_mismatch",
"index": entry_id,
"kind": kind,
"expected_kind": EntryType.DOCUMENT.value,
}
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
token_meta: dict[str, Any] | None = None
if raw_token:
ok, token_reason, token_meta = _validate_token(
raw_token,
operation="export",
kind=kind,
label=label,
consume_use=True,
)
if not ok:
payload = {
"status": "denied",
"reason": token_reason,
"kind": kind,
"label": label,
"index": entry_id,
}
_append_audit_event(
"agent_document_export_denied",
{
"fingerprint": fingerprint,
"index": entry_id,
"kind": kind,
"label": label,
"reason": token_reason,
},
)
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
if approval_required(policy, "export"):
if not approval_id:
payload = {
"status": "denied",
"reason": "policy_deny:approval_required",
"action": "export",
"kind": kind,
"label": label,
"index": entry_id,
}
_append_audit_event(
"agent_document_export_denied",
{
"fingerprint": fingerprint,
"index": entry_id,
"kind": kind,
"label": label,
"reason": "policy_deny:approval_required",
"action": "export",
},
)
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
ok, approval_reason = consume_approval(
approval_id=approval_id,
action="export",
resource=f"entry:{kind}:{entry_id}",
)
if not ok:
reason = f"policy_deny:{approval_reason}"
payload = {
"status": "denied",
"reason": reason,
"action": "export",
"kind": kind,
"label": label,
"index": entry_id,
}
_append_audit_event(
"agent_document_export_denied",
{
"fingerprint": fingerprint,
"index": entry_id,
"kind": kind,
"label": label,
"reason": reason,
"action": "export",
"approval_id": approval_id,
},
)
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
try:
destination = service.export_document_file(entry_id, out, overwrite=overwrite)
except Exception as exc:
payload = {
"status": "error",
"reason": "document_export_failed",
"detail": str(exc),
"index": entry_id,
}
typer.echo(json.dumps(payload, indent=2))
raise typer.Exit(1)
payload = {
"status": "ok",
"fingerprint": fingerprint,
"index": entry_id,
"kind": kind,
"label": label,
"output_path": str(destination),
"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_document_export_granted",
{
"fingerprint": fingerprint,
"index": entry_id,
"kind": kind,
"label": label,
"output_path": str(destination),
"token_id": (token_meta or {}).get("id"),
},
)
typer.echo(json.dumps(payload, indent=2))
@app.command("get") @app.command("get")
def agent_get( def agent_get(
ctx: typer.Context, ctx: typer.Context,
+29
View File
@@ -38,6 +38,7 @@ def build_capabilities() -> dict[str, Any]:
"/docs", "/docs",
"/api/v1/export/check", "/api/v1/export/check",
"/api/v1/export/manifest/verify", "/api/v1/export/manifest/verify",
"/api/v1/entry/{id}/links",
"/api/v1/agent/job-profiles", "/api/v1/agent/job-profiles",
"/api/v1/agent/job-profiles/{job_id}/run", "/api/v1/agent/job-profiles/{job_id}/run",
"/api/v1/agent/job-profiles/{job_id}/template", "/api/v1/agent/job-profiles/{job_id}/template",
@@ -180,6 +181,30 @@ def build_capabilities() -> dict[str, Any]:
"supports_manifest_entry_hash_verification": True, "supports_manifest_entry_hash_verification": True,
"manifest_mode": "policy_filtered", "manifest_mode": "policy_filtered",
}, },
"document_io": {
"commands": [
"entry import-document",
"entry export-document",
"agent document-import",
"agent document-export",
],
"requires_agent_export_import_policy": True,
},
"knowledge_graph": {
"commands": [
"entry link-add",
"entry links",
"entry link-remove",
"entry modify --links-json",
],
"api_endpoints": [
"GET /api/v1/entry/{id}/links",
"POST /api/v1/entry/{id}/links",
"DELETE /api/v1/entry/{id}/links",
],
"search_includes_link_metadata": True,
"sync_union_merge_fields": ["tags", "custom_fields", "links"],
},
"redaction": { "redaction": {
"default_safe_output": True, "default_safe_output": True,
"reveal_override": "agent get --reveal", "reveal_override": "agent get --reveal",
@@ -214,6 +239,8 @@ def build_capabilities() -> dict[str, Any]:
"Run `seedpass --help` for top-level commands.", "Run `seedpass --help` for top-level commands.",
"Run `seedpass <group> --help` for detailed options.", "Run `seedpass <group> --help` for detailed options.",
"Use `seedpass capabilities --format json` for machine-readable discovery.", "Use `seedpass capabilities --format json` for machine-readable discovery.",
"For graph workflows, start with `seedpass entry link-add --help` and `seedpass entry links --help`.",
"For document workflows, run `seedpass entry import-document --help` and `seedpass agent document-import --help`.",
"After unlock/login, run `seedpass capabilities` before autonomous runs.", "After unlock/login, run `seedpass capabilities` before autonomous runs.",
"For agent bootstrap context, run `seedpass agent bootstrap-context`.", "For agent bootstrap context, run `seedpass agent bootstrap-context`.",
], ],
@@ -252,6 +279,8 @@ def emit_capabilities(output_format: str) -> str:
lines.append("- Safe output redaction default with explicit --reveal") lines.append("- Safe output redaction default with explicit --reveal")
lines.append("- Security posture checks and remediation bundles") lines.append("- Security posture checks and remediation bundles")
lines.append("- Audited access log with chained HMAC signatures") lines.append("- Audited access log with chained HMAC signatures")
lines.append("- Knowledge graph links: entry link-add/links/link-remove")
lines.append("- Document file workflows for users and agents")
lines.append("") lines.append("")
lines.append("Discovery hints:") lines.append("Discovery hints:")
for hint in data["help_hints"]: for hint in data["help_hints"]:
+128
View File
@@ -108,6 +108,8 @@ def entry_get(ctx: typer.Context, query: str) -> None:
elif etype == EntryType.TOTP.value: elif etype == EntryType.TOTP.value:
code = service.get_totp_code(index) code = service.get_totp_code(index)
typer.echo(code) typer.echo(code)
elif etype == EntryType.DOCUMENT.value:
typer.echo(str(entry.get("content", "")))
else: else:
typer.echo("Unsupported entry type") typer.echo("Unsupported entry type")
raise typer.Exit(code=1) raise typer.Exit(code=1)
@@ -302,6 +304,55 @@ def entry_add_managed_account(
typer.echo(str(idx)) typer.echo(str(idx))
@app.command("add-document")
def entry_add_document(
ctx: typer.Context,
label: str,
content: str = typer.Option("", "--content", help="Document text content"),
file_type: str = typer.Option("txt", "--file-type", help="File extension/type"),
notes: str = typer.Option("", "--notes", help="Entry notes"),
) -> None:
"""Add a document entry and output its index."""
service = _get_entry_service(ctx)
idx = service.add_document(
label,
content,
file_type=file_type,
notes=notes,
)
typer.echo(str(idx))
@app.command("import-document")
def entry_import_document(
ctx: typer.Context,
file: str = typer.Option(..., "--file", help="Input file path"),
label: Optional[str] = typer.Option(None, "--label", help="Title override"),
notes: str = typer.Option("", "--notes", help="Entry notes"),
) -> None:
"""Import a local text file as a document entry and output its index."""
service = _get_entry_service(ctx)
idx = service.import_document_file(file, label=label, notes=notes)
typer.echo(str(idx))
@app.command("export-document")
def entry_export_document(
ctx: typer.Context,
entry_id: int = typer.Option(..., "--entry-id", help="Document entry index"),
out: Optional[str] = typer.Option(
None,
"--out",
help="Output file path or directory (default: current directory)",
),
overwrite: bool = typer.Option(False, "--overwrite", help="Overwrite if exists"),
) -> None:
"""Export a document entry to a local file and print the destination path."""
service = _get_entry_service(ctx)
dest = service.export_document_file(entry_id, out, overwrite=overwrite)
typer.echo(str(dest))
@app.command("modify") @app.command("modify")
def entry_modify( def entry_modify(
ctx: typer.Context, ctx: typer.Context,
@@ -316,9 +367,31 @@ def entry_modify(
digits: Optional[int] = typer.Option(None, "--digits", help="TOTP digits"), digits: Optional[int] = typer.Option(None, "--digits", help="TOTP digits"),
key: Optional[str] = typer.Option(None, "--key", help="New key"), key: Optional[str] = typer.Option(None, "--key", help="New key"),
value: Optional[str] = typer.Option(None, "--value", help="New value"), value: Optional[str] = typer.Option(None, "--value", help="New value"),
content: Optional[str] = typer.Option(
None, "--content", help="New document content"
),
file_type: Optional[str] = typer.Option(
None, "--file-type", help="New document file type"
),
links_json: Optional[str] = typer.Option(
None,
"--links-json",
help="Replace links with JSON array: [{\"target_id\":1,\"relation\":\"related_to\",\"note\":\"...\"}]",
),
) -> None: ) -> None:
"""Modify an existing entry.""" """Modify an existing entry."""
service = _get_entry_service(ctx) service = _get_entry_service(ctx)
links = None
if links_json is not None:
try:
parsed = json.loads(links_json)
except json.JSONDecodeError as exc:
typer.echo(f"Invalid --links-json: {exc}")
raise typer.Exit(code=1)
if not isinstance(parsed, list):
typer.echo("Invalid --links-json: expected a JSON list")
raise typer.Exit(code=1)
links = parsed
try: try:
service.modify_entry( service.modify_entry(
entry_id, entry_id,
@@ -330,6 +403,9 @@ def entry_modify(
digits=digits, digits=digits,
key=key, key=key,
value=value, value=value,
content=content,
file_type=file_type,
links=links,
) )
except ValueError as e: except ValueError as e:
typer.echo(str(e)) typer.echo(str(e))
@@ -337,6 +413,58 @@ def entry_modify(
raise typer.Exit(code=1) raise typer.Exit(code=1)
@app.command("links")
def entry_links(
ctx: typer.Context,
entry_id: int = typer.Option(..., "--entry-id", help="Entry index"),
) -> None:
"""List graph links for an entry."""
service = _get_entry_service(ctx)
try:
links = service.get_links(entry_id)
except ValueError as exc:
typer.echo(str(exc))
raise typer.Exit(code=1)
typer.echo(json.dumps({"entry_id": entry_id, "links": links}, indent=2))
@app.command("link-add")
def entry_link_add(
ctx: typer.Context,
entry_id: int = typer.Option(..., "--entry-id", help="Source entry index"),
target_id: int = typer.Option(..., "--target-id", help="Target entry index"),
relation: str = typer.Option("related_to", "--relation", help="Link relation"),
note: str = typer.Option("", "--note", help="Optional link note"),
) -> None:
"""Create a typed link between entries."""
service = _get_entry_service(ctx)
try:
links = service.add_link(entry_id, target_id, relation=relation, note=note)
except ValueError as exc:
typer.echo(str(exc))
raise typer.Exit(code=1)
typer.echo(json.dumps({"entry_id": entry_id, "links": links}, indent=2))
@app.command("link-remove")
def entry_link_remove(
ctx: typer.Context,
entry_id: int = typer.Option(..., "--entry-id", help="Source entry index"),
target_id: int = typer.Option(..., "--target-id", help="Target entry index"),
relation: Optional[str] = typer.Option(
None, "--relation", help="Relation filter (optional)"
),
) -> None:
"""Remove links from source entry to target entry."""
service = _get_entry_service(ctx)
try:
links = service.remove_link(entry_id, target_id, relation=relation)
except ValueError as exc:
typer.echo(str(exc))
raise typer.Exit(code=1)
typer.echo(json.dumps({"entry_id": entry_id, "links": links}, indent=2))
@app.command("archive") @app.command("archive")
def entry_archive(ctx: typer.Context, entry_id: int) -> None: def entry_archive(ctx: typer.Context, entry_id: int) -> None:
"""Archive an entry.""" """Archive an entry."""
+1
View File
@@ -18,6 +18,7 @@ DEFAULT_EXPORT_POLICY: dict[str, Any] = {
EntryType.PASSWORD.value, EntryType.PASSWORD.value,
EntryType.TOTP.value, EntryType.TOTP.value,
EntryType.KEY_VALUE.value, EntryType.KEY_VALUE.value,
EntryType.DOCUMENT.value,
], ],
"allow_export_import": False, "allow_export_import": False,
"export": {"allow_full_vault": False}, "export": {"allow_full_vault": False},
+101
View File
@@ -487,6 +487,62 @@ class EntryService:
self._manager.start_background_vault_sync() self._manager.start_background_vault_sync()
return idx return idx
def add_document(
self,
label: str,
content: str,
*,
file_type: str = "txt",
notes: str = "",
tags: list[str] | None = None,
archived: bool = False,
) -> int:
with self._lock:
idx = self._manager.entry_manager.add_document(
label,
content,
file_type=file_type,
notes=notes,
tags=tags,
archived=archived,
)
self._manager.start_background_vault_sync()
return idx
def import_document_file(
self,
file_path: str | Path,
*,
label: str | None = None,
notes: str = "",
tags: list[str] | None = None,
archived: bool = False,
) -> int:
with self._lock:
idx = self._manager.entry_manager.import_document_file(
file_path,
label=label,
notes=notes,
tags=tags,
archived=archived,
)
self._manager.start_background_vault_sync()
return idx
def export_document_file(
self,
entry_id: int,
output_path: str | Path | None = None,
*,
overwrite: bool = False,
) -> Path:
with self._lock:
return self._manager.entry_manager.export_document_file(
entry_id,
output_path=output_path,
overwrite=overwrite,
)
def modify_entry( def modify_entry(
self, self,
entry_id: int, entry_id: int,
@@ -499,6 +555,12 @@ class EntryService:
digits: int | None = None, digits: int | None = None,
key: str | None = None, key: str | None = None,
value: str | None = None, value: str | None = None,
content: str | None = None,
file_type: str | None = None,
custom_fields: list[dict[str, Any]] | None = None,
tags: list[str] | None = None,
links: list[dict[str, Any]] | None = None,
archived: bool | None = None,
) -> None: ) -> None:
with self._lock: with self._lock:
self._manager.entry_manager.modify_entry( self._manager.entry_manager.modify_entry(
@@ -511,9 +573,48 @@ class EntryService:
digits=digits, digits=digits,
key=key, key=key,
value=value, value=value,
content=content,
file_type=file_type,
custom_fields=custom_fields,
tags=tags,
links=links,
archived=archived,
) )
self._manager.start_background_vault_sync() self._manager.start_background_vault_sync()
def add_link(
self,
entry_id: int,
target_id: int,
*,
relation: str = "related_to",
note: str = "",
) -> list[dict[str, Any]]:
with self._lock:
links = self._manager.entry_manager.add_link(
entry_id, target_id, relation=relation, note=note
)
self._manager.start_background_vault_sync()
return links
def remove_link(
self,
entry_id: int,
target_id: int,
*,
relation: str | None = None,
) -> list[dict[str, Any]]:
with self._lock:
links = self._manager.entry_manager.remove_link(
entry_id, target_id, relation=relation
)
self._manager.start_background_vault_sync()
return links
def get_links(self, entry_id: int) -> list[dict[str, Any]]:
with self._lock:
return self._manager.entry_manager.get_links(entry_id)
def archive_entry(self, entry_id: int) -> None: def archive_entry(self, entry_id: int) -> None:
with self._lock: with self._lock:
self._manager.entry_manager.archive_entry(entry_id) self._manager.entry_manager.archive_entry(entry_id)
+47
View File
@@ -69,6 +69,8 @@ class DisplayService:
self._display_nostr(entry, index) self._display_nostr(entry, index)
elif entry_type == EntryType.KEY_VALUE.value: elif entry_type == EntryType.KEY_VALUE.value:
self._display_key_value(entry, index) self._display_key_value(entry, index)
elif entry_type == EntryType.DOCUMENT.value:
self._display_document(entry, index)
elif entry_type == EntryType.MANAGED_ACCOUNT.value: elif entry_type == EntryType.MANAGED_ACCOUNT.value:
self._display_managed_account(entry, index) self._display_managed_account(entry, index)
else: else:
@@ -382,6 +384,51 @@ class DisplayService:
self.manager.load_managed_account(index) self.manager.load_managed_account(index)
return return
def _display_document(self, entry: dict, _index: int) -> None:
label = entry.get("label", "")
file_type = entry.get("file_type", "txt")
notes = entry.get("notes", "")
content = str(entry.get("content", ""))
archived = entry.get("archived", False)
print(colored(f"Document '{label}' ({file_type})", "cyan"))
if notes:
print(colored(f"Notes: {notes}", "cyan"))
tags = entry.get("tags", [])
if tags:
print(colored(f"Tags: {', '.join(tags)}", "cyan"))
print(
colored(f"Archived Status: {'Archived' if archived else 'Active'}", "cyan")
)
print(color_text("\n--- BEGIN DOCUMENT ---", "index"))
print(color_text(content, "index"))
print(color_text("--- END DOCUMENT ---\n", "index"))
custom_fields = entry.get("custom_fields", [])
if custom_fields:
print(colored("Additional Fields:", "cyan"))
hidden_fields = []
for field in custom_fields:
f_label = field.get("label", "")
f_value = field.get("value", "")
if field.get("is_hidden"):
hidden_fields.append((f_label, f_value))
print(colored(f" {f_label}: [hidden]", "cyan"))
else:
print(colored(f" {f_label}: {f_value}", "cyan"))
if hidden_fields:
show = input("Reveal hidden fields? (y/N): ").strip().lower()
if show == "y":
for f_label, f_value in hidden_fields:
if self.manager.secret_mode_enabled:
if self._copy_to_clipboard(f_value):
print(
colored(
f"[+] {f_label} copied to clipboard. Will clear in {self.manager.clipboard_clear_delay} seconds.",
"green",
)
)
else:
print(colored(f" {f_label}: {f_value}", "cyan"))
def _display_password(self, entry: dict, index: int) -> None: def _display_password(self, entry: dict, index: int) -> None:
website_name = entry.get("label", entry.get("website")) website_name = entry.get("label", entry.get("website"))
length = entry.get("length") length = entry.get("length")
+420 -24
View File
@@ -25,8 +25,10 @@ except Exception: # pragma: no cover - fallback when orjson is missing
USE_ORJSON = False USE_ORJSON = False
import logging import logging
import hashlib import hashlib
import re
import shutil import shutil
import time import time
from datetime import datetime, timezone
from typing import Optional, Tuple, Dict, Any, List from typing import Optional, Tuple, Dict, Any, List
from pathlib import Path from pathlib import Path
@@ -77,15 +79,46 @@ class EntryManager:
"""Clear the cached index data.""" """Clear the cached index data."""
self._index_cache = None self._index_cache = None
def _load_index(self, force_reload: bool = False) -> Dict[str, Any]: @staticmethod
if not force_reload and self._index_cache is not None: def _now_unix() -> int:
return self._index_cache return int(time.time())
if self.index_file.exists(): @staticmethod
def _iso_from_unix(ts: int) -> str:
return datetime.fromtimestamp(int(ts), tz=timezone.utc).isoformat()
def _now_iso(self) -> str:
return self._iso_from_unix(self._now_unix())
@staticmethod
def _normalize_links(raw_links: Any) -> list[dict[str, Any]]:
"""Return canonical, de-duplicated entry links."""
if not isinstance(raw_links, list):
return []
normalized: list[dict[str, Any]] = []
seen: set[tuple[int, str, str]] = set()
for item in raw_links:
if not isinstance(item, dict):
continue
try: try:
data = self.vault.load_index() target_id = int(item.get("target_id"))
# Normalize legacy fields except Exception:
for entry in data.get("entries", {}).values(): continue
relation = str(item.get("relation", "related_to")).strip().lower()
note = str(item.get("note", "")).strip()
if target_id < 0 or not relation:
continue
key = (target_id, relation, note)
if key in seen:
continue
seen.add(key)
normalized.append(
{"target_id": target_id, "relation": relation, "note": note}
)
return normalized
def _normalize_entry_defaults(self, entry: dict[str, Any]) -> dict[str, Any]:
"""Normalize legacy fields and fill default graph/meta fields."""
if "type" not in entry and "kind" in entry: if "type" not in entry and "kind" in entry:
entry["type"] = entry["kind"] entry["type"] = entry["kind"]
if "kind" not in entry: if "kind" not in entry:
@@ -93,10 +126,7 @@ class EntryManager:
entry.setdefault("type", entry["kind"]) entry.setdefault("type", entry["kind"])
if "label" not in entry and "website" in entry: if "label" not in entry and "website" in entry:
entry["label"] = entry["website"] entry["label"] = entry["website"]
if ( if "website" in entry and entry.get("type") == EntryType.PASSWORD.value:
"website" in entry
and entry.get("type") == EntryType.PASSWORD.value
):
entry.pop("website", None) entry.pop("website", None)
if "archived" not in entry and "blacklisted" in entry: if "archived" not in entry and "blacklisted" in entry:
entry["archived"] = entry["blacklisted"] entry["archived"] = entry["blacklisted"]
@@ -106,6 +136,40 @@ class EntryManager:
entry.pop("words", None) entry.pop("words", None)
entry.setdefault("tags", []) entry.setdefault("tags", [])
entry.setdefault("modified_ts", entry.get("updated", 0)) entry.setdefault("modified_ts", entry.get("updated", 0))
modified_ts = int(entry.get("modified_ts", 0) or 0)
if "date_added" not in entry:
entry["date_added"] = self._iso_from_unix(modified_ts)
if "date_modified" not in entry:
entry["date_modified"] = self._iso_from_unix(modified_ts)
entry["links"] = self._normalize_links(entry.get("links", []))
if entry.get("type") == EntryType.DOCUMENT.value:
entry.setdefault("content", "")
entry.setdefault("file_type", "txt")
entry.setdefault("custom_fields", [])
if entry.get("type") in (
EntryType.PASSWORD.value,
EntryType.KEY_VALUE.value,
EntryType.MANAGED_ACCOUNT.value,
EntryType.DOCUMENT.value,
EntryType.SSH.value,
):
entry.setdefault("custom_fields", [])
return entry
def _load_index(self, force_reload: bool = False) -> Dict[str, Any]:
if not force_reload and self._index_cache is not None:
for entry in self._index_cache.get("entries", {}).values():
if isinstance(entry, dict):
self._normalize_entry_defaults(entry)
return self._index_cache
if self.index_file.exists():
try:
data = self.vault.load_index()
# Normalize legacy fields
for entry in data.get("entries", {}).values():
if isinstance(entry, dict):
self._normalize_entry_defaults(entry)
logger.debug("Index loaded successfully.") logger.debug("Index loaded successfully.")
self._index_cache = data self._index_cache = data
return data return data
@@ -183,6 +247,8 @@ class EntryManager:
try: try:
index = self.get_next_index() index = self.get_next_index()
data = self._load_index() data = self._load_index()
now_unix = self._now_unix()
now_iso = self._iso_from_unix(now_unix)
data.setdefault("entries", {}) data.setdefault("entries", {})
entry = { entry = {
@@ -194,9 +260,12 @@ class EntryManager:
"type": EntryType.PASSWORD.value, "type": EntryType.PASSWORD.value,
"kind": EntryType.PASSWORD.value, "kind": EntryType.PASSWORD.value,
"notes": notes, "notes": notes,
"modified_ts": int(time.time()), "modified_ts": now_unix,
"date_added": now_iso,
"date_modified": now_iso,
"custom_fields": custom_fields or [], "custom_fields": custom_fields or [],
"tags": tags or [], "tags": tags or [],
"links": [],
} }
policy: dict[str, Any] = {} policy: dict[str, Any] = {}
@@ -271,6 +340,8 @@ class EntryManager:
entry_id = self.get_next_index() entry_id = self.get_next_index()
data = self._load_index() data = self._load_index()
data.setdefault("entries", {}) data.setdefault("entries", {})
now_unix = self._now_unix()
now_iso = self._iso_from_unix(now_unix)
if deterministic: if deterministic:
if parent_seed is None: if parent_seed is None:
@@ -284,7 +355,9 @@ class EntryManager:
"type": EntryType.TOTP.value, "type": EntryType.TOTP.value,
"kind": EntryType.TOTP.value, "kind": EntryType.TOTP.value,
"label": label, "label": label,
"modified_ts": int(time.time()), "modified_ts": now_unix,
"date_added": now_iso,
"date_modified": now_iso,
"index": index, "index": index,
"period": period, "period": period,
"digits": digits, "digits": digits,
@@ -292,6 +365,7 @@ class EntryManager:
"notes": notes, "notes": notes,
"tags": tags or [], "tags": tags or [],
"deterministic": True, "deterministic": True,
"links": [],
} }
else: else:
if secret is None: if secret is None:
@@ -303,13 +377,16 @@ class EntryManager:
"kind": EntryType.TOTP.value, "kind": EntryType.TOTP.value,
"label": label, "label": label,
"secret": secret, "secret": secret,
"modified_ts": int(time.time()), "modified_ts": now_unix,
"date_added": now_iso,
"date_modified": now_iso,
"period": period, "period": period,
"digits": digits, "digits": digits,
"archived": archived, "archived": archived,
"notes": notes, "notes": notes,
"tags": tags or [], "tags": tags or [],
"deterministic": False, "deterministic": False,
"links": [],
} }
data["entries"][str(entry_id)] = entry data["entries"][str(entry_id)] = entry
@@ -348,6 +425,8 @@ class EntryManager:
priv_pem, pub_pem = derive_ssh_key_pair(parent_seed, index) priv_pem, pub_pem = derive_ssh_key_pair(parent_seed, index)
if not validate_ssh_key_pair(priv_pem, pub_pem): if not validate_ssh_key_pair(priv_pem, pub_pem):
raise ValueError("Derived SSH key pair failed validation") raise ValueError("Derived SSH key pair failed validation")
now_unix = self._now_unix()
now_iso = self._iso_from_unix(now_unix)
data = self._load_index() data = self._load_index()
data.setdefault("entries", {}) data.setdefault("entries", {})
@@ -356,10 +435,14 @@ class EntryManager:
"kind": EntryType.SSH.value, "kind": EntryType.SSH.value,
"index": index, "index": index,
"label": label, "label": label,
"modified_ts": int(time.time()), "modified_ts": now_unix,
"date_added": now_iso,
"date_modified": now_iso,
"notes": notes, "notes": notes,
"archived": archived, "archived": archived,
"tags": tags or [], "tags": tags or [],
"custom_fields": [],
"links": [],
} }
self._save_index(data) self._save_index(data)
self.update_checksum() self.update_checksum()
@@ -406,6 +489,8 @@ class EntryManager:
priv_key, fp = derive_pgp_key(bip85, index, key_type, user_id) priv_key, fp = derive_pgp_key(bip85, index, key_type, user_id)
if not validate_pgp_private_key(priv_key, fp): if not validate_pgp_private_key(priv_key, fp):
raise ValueError("Derived PGP key failed validation") raise ValueError("Derived PGP key failed validation")
now_unix = self._now_unix()
now_iso = self._iso_from_unix(now_unix)
data = self._load_index() data = self._load_index()
data.setdefault("entries", {}) data.setdefault("entries", {})
@@ -414,12 +499,15 @@ class EntryManager:
"kind": EntryType.PGP.value, "kind": EntryType.PGP.value,
"index": index, "index": index,
"label": label, "label": label,
"modified_ts": int(time.time()), "modified_ts": now_unix,
"date_added": now_iso,
"date_modified": now_iso,
"key_type": key_type, "key_type": key_type,
"user_id": user_id, "user_id": user_id,
"notes": notes, "notes": notes,
"archived": archived, "archived": archived,
"tags": tags or [], "tags": tags or [],
"links": [],
} }
self._save_index(data) self._save_index(data)
self.update_checksum() self.update_checksum()
@@ -473,6 +561,8 @@ class EntryManager:
nsec = Keys.hex_to_bech32(keys.private_key_hex(), "nsec") nsec = Keys.hex_to_bech32(keys.private_key_hex(), "nsec")
if not validate_nostr_keys(npub, nsec): if not validate_nostr_keys(npub, nsec):
raise ValueError("Derived Nostr keys failed validation") raise ValueError("Derived Nostr keys failed validation")
now_unix = self._now_unix()
now_iso = self._iso_from_unix(now_unix)
data = self._load_index() data = self._load_index()
data.setdefault("entries", {}) data.setdefault("entries", {})
@@ -481,10 +571,13 @@ class EntryManager:
"kind": EntryType.NOSTR.value, "kind": EntryType.NOSTR.value,
"index": index, "index": index,
"label": label, "label": label,
"modified_ts": int(time.time()), "modified_ts": now_unix,
"date_added": now_iso,
"date_modified": now_iso,
"notes": notes, "notes": notes,
"archived": archived, "archived": archived,
"tags": tags or [], "tags": tags or [],
"links": [],
} }
self._save_index(data) self._save_index(data)
self.update_checksum() self.update_checksum()
@@ -505,6 +598,8 @@ class EntryManager:
"""Add a new generic key/value entry.""" """Add a new generic key/value entry."""
index = self.get_next_index() index = self.get_next_index()
now_unix = self._now_unix()
now_iso = self._iso_from_unix(now_unix)
data = self._load_index() data = self._load_index()
data.setdefault("entries", {}) data.setdefault("entries", {})
@@ -513,12 +608,15 @@ class EntryManager:
"kind": EntryType.KEY_VALUE.value, "kind": EntryType.KEY_VALUE.value,
"label": label, "label": label,
"key": key, "key": key,
"modified_ts": int(time.time()), "modified_ts": now_unix,
"date_added": now_iso,
"date_modified": now_iso,
"value": value, "value": value,
"notes": notes, "notes": notes,
"archived": archived, "archived": archived,
"custom_fields": custom_fields or [], "custom_fields": custom_fields or [],
"tags": tags or [], "tags": tags or [],
"links": [],
} }
self._save_index(data) self._save_index(data)
@@ -526,6 +624,111 @@ class EntryManager:
self.backup_manager.create_backup() self.backup_manager.create_backup()
return index return index
def add_document(
self,
label: str,
content: str,
file_type: str = "txt",
*,
notes: str = "",
archived: bool = False,
tags: list[str] | None = None,
) -> int:
"""Add a new free-form document entry."""
index = self.get_next_index()
now_unix = self._now_unix()
now_iso = self._iso_from_unix(now_unix)
data = self._load_index()
data.setdefault("entries", {})
data["entries"][str(index)] = {
"type": EntryType.DOCUMENT.value,
"kind": EntryType.DOCUMENT.value,
"label": label,
"content": content,
"file_type": (file_type or "txt").strip().lower(),
"modified_ts": now_unix,
"date_added": now_iso,
"date_modified": now_iso,
"notes": notes,
"archived": archived,
"tags": tags or [],
"links": [],
}
self._save_index(data)
self.update_checksum()
self.backup_manager.create_backup()
return index
def import_document_file(
self,
file_path: str | Path,
*,
label: str | None = None,
notes: str = "",
archived: bool = False,
tags: list[str] | None = None,
) -> int:
"""Import a plaintext document file as a document entry."""
path = Path(file_path).expanduser()
if not path.exists() or not path.is_file():
raise FileNotFoundError(f"Document file not found: {path}")
content = path.read_text(encoding="utf-8")
file_type = path.suffix.lstrip(".").lower() or "txt"
entry_label = label if label else path.stem
return self.add_document(
entry_label,
content,
file_type=file_type,
notes=notes,
archived=archived,
tags=tags,
)
def export_document_file(
self,
index: int,
output_path: str | Path | None = None,
*,
overwrite: bool = False,
) -> Path:
"""Export a document entry to a plaintext file on disk."""
entry = self.retrieve_entry(index)
if not entry:
raise ValueError(f"Entry not found: {index}")
kind = str(entry.get("type", entry.get("kind", ""))).strip().lower()
if kind != EntryType.DOCUMENT.value:
raise ValueError("Entry is not a document entry")
file_type = str(entry.get("file_type", "txt")).strip().lower() or "txt"
label = str(entry.get("label", f"document_{index}")).strip()
stem = (
re.sub(r"[^A-Za-z0-9._-]+", "_", label).strip("._") or f"document_{index}"
)
content = str(entry.get("content", ""))
if output_path is None:
dest = Path.cwd() / f"{stem}.{file_type}"
else:
raw = Path(output_path).expanduser()
if raw.exists() and raw.is_dir():
dest = raw / f"{stem}.{file_type}"
elif raw.suffix:
dest = raw
else:
dest = raw / f"{stem}.{file_type}"
dest.parent.mkdir(parents=True, exist_ok=True)
if dest.exists() and not overwrite:
raise FileExistsError(f"File already exists: {dest}")
atomic_write(dest, lambda f: f.write(content))
return dest
def get_nostr_key_pair(self, index: int, parent_seed: str) -> tuple[str, str]: def get_nostr_key_pair(self, index: int, parent_seed: str) -> tuple[str, str]:
"""Return the npub and nsec for the specified entry.""" """Return the npub and nsec for the specified entry."""
@@ -575,6 +778,8 @@ class EntryManager:
phrase = derive_seed_phrase(bip85, index, words_num) phrase = derive_seed_phrase(bip85, index, words_num)
if not validate_seed_phrase(phrase): if not validate_seed_phrase(phrase):
raise ValueError("Derived seed phrase failed validation") raise ValueError("Derived seed phrase failed validation")
now_unix = self._now_unix()
now_iso = self._iso_from_unix(now_unix)
data = self._load_index() data = self._load_index()
data.setdefault("entries", {}) data.setdefault("entries", {})
@@ -583,11 +788,14 @@ class EntryManager:
"kind": EntryType.SEED.value, "kind": EntryType.SEED.value,
"index": index, "index": index,
"label": label, "label": label,
"modified_ts": int(time.time()), "modified_ts": now_unix,
"date_added": now_iso,
"date_modified": now_iso,
"word_count": words_num, "word_count": words_num,
"notes": notes, "notes": notes,
"archived": archived, "archived": archived,
"tags": tags or [], "tags": tags or [],
"links": [],
} }
self._save_index(data) self._save_index(data)
self.update_checksum() self.update_checksum()
@@ -647,6 +855,8 @@ class EntryManager:
if not validate_seed_phrase(seed_phrase): if not validate_seed_phrase(seed_phrase):
raise ValueError("Derived managed account seed failed validation") raise ValueError("Derived managed account seed failed validation")
fingerprint = generate_fingerprint(seed_phrase) fingerprint = generate_fingerprint(seed_phrase)
now_unix = self._now_unix()
now_iso = self._iso_from_unix(now_unix)
account_dir = self.fingerprint_dir / "accounts" / fingerprint account_dir = self.fingerprint_dir / "accounts" / fingerprint
account_dir.mkdir(parents=True, exist_ok=True) account_dir.mkdir(parents=True, exist_ok=True)
@@ -658,12 +868,15 @@ class EntryManager:
"kind": EntryType.MANAGED_ACCOUNT.value, "kind": EntryType.MANAGED_ACCOUNT.value,
"index": index, "index": index,
"label": label, "label": label,
"modified_ts": int(time.time()), "modified_ts": now_unix,
"date_added": now_iso,
"date_modified": now_iso,
"word_count": word_count, "word_count": word_count,
"notes": notes, "notes": notes,
"fingerprint": fingerprint, "fingerprint": fingerprint,
"archived": archived, "archived": archived,
"tags": tags or [], "tags": tags or [],
"links": [],
} }
self._save_index(data) self._save_index(data)
@@ -813,6 +1026,7 @@ class EntryManager:
EntryType.PASSWORD.value, EntryType.PASSWORD.value,
EntryType.KEY_VALUE.value, EntryType.KEY_VALUE.value,
EntryType.MANAGED_ACCOUNT.value, EntryType.MANAGED_ACCOUNT.value,
EntryType.DOCUMENT.value,
): ):
entry.setdefault("custom_fields", []) entry.setdefault("custom_fields", [])
logger.debug( logger.debug(
@@ -847,8 +1061,11 @@ class EntryManager:
digits: Optional[int] = None, digits: Optional[int] = None,
key: Optional[str] = None, key: Optional[str] = None,
value: Optional[str] = None, value: Optional[str] = None,
content: Optional[str] = None,
file_type: Optional[str] = None,
custom_fields: List[Dict[str, Any]] | None = None, custom_fields: List[Dict[str, Any]] | None = None,
tags: list[str] | None = None, tags: list[str] | None = None,
links: List[Dict[str, Any]] | None = None,
include_special_chars: bool | None = None, include_special_chars: bool | None = None,
allowed_special_chars: str | None = None, allowed_special_chars: str | None = None,
special_mode: str | None = None, special_mode: str | None = None,
@@ -901,8 +1118,11 @@ class EntryManager:
"digits": digits, "digits": digits,
"key": key, "key": key,
"value": value, "value": value,
"content": content,
"file_type": file_type,
"custom_fields": custom_fields, "custom_fields": custom_fields,
"tags": tags, "tags": tags,
"links": links,
"include_special_chars": include_special_chars, "include_special_chars": include_special_chars,
"allowed_special_chars": allowed_special_chars, "allowed_special_chars": allowed_special_chars,
"special_mode": special_mode, "special_mode": special_mode,
@@ -922,6 +1142,7 @@ class EntryManager:
"notes", "notes",
"custom_fields", "custom_fields",
"tags", "tags",
"links",
"include_special_chars", "include_special_chars",
"allowed_special_chars", "allowed_special_chars",
"special_mode", "special_mode",
@@ -939,6 +1160,7 @@ class EntryManager:
"notes", "notes",
"custom_fields", "custom_fields",
"tags", "tags",
"links",
}, },
EntryType.KEY_VALUE.value: { EntryType.KEY_VALUE.value: {
"label", "label",
@@ -948,6 +1170,7 @@ class EntryManager:
"notes", "notes",
"custom_fields", "custom_fields",
"tags", "tags",
"links",
}, },
EntryType.MANAGED_ACCOUNT.value: { EntryType.MANAGED_ACCOUNT.value: {
"label", "label",
@@ -956,6 +1179,17 @@ class EntryManager:
"notes", "notes",
"custom_fields", "custom_fields",
"tags", "tags",
"links",
},
EntryType.DOCUMENT.value: {
"label",
"content",
"file_type",
"archived",
"notes",
"tags",
"custom_fields",
"links",
}, },
EntryType.SSH.value: { EntryType.SSH.value: {
"label", "label",
@@ -963,6 +1197,7 @@ class EntryManager:
"notes", "notes",
"custom_fields", "custom_fields",
"tags", "tags",
"links",
}, },
EntryType.PGP.value: { EntryType.PGP.value: {
"label", "label",
@@ -970,6 +1205,7 @@ class EntryManager:
"notes", "notes",
"custom_fields", "custom_fields",
"tags", "tags",
"links",
}, },
EntryType.NOSTR.value: { EntryType.NOSTR.value: {
"label", "label",
@@ -977,6 +1213,7 @@ class EntryManager:
"notes", "notes",
"custom_fields", "custom_fields",
"tags", "tags",
"links",
}, },
EntryType.SEED.value: { EntryType.SEED.value: {
"label", "label",
@@ -984,6 +1221,7 @@ class EntryManager:
"notes", "notes",
"custom_fields", "custom_fields",
"tags", "tags",
"links",
}, },
} }
@@ -1029,6 +1267,13 @@ class EntryManager:
if value is not None: if value is not None:
entry["value"] = value entry["value"] = value
logger.debug(f"Updated value for index {index}.") logger.debug(f"Updated value for index {index}.")
elif entry_type == EntryType.DOCUMENT.value:
if content is not None:
entry["content"] = content
logger.debug(f"Updated document content for index {index}.")
if file_type is not None:
entry["file_type"] = file_type.strip().lower()
logger.debug(f"Updated document file type for index {index}.")
if archived is None and "blacklisted" in legacy: if archived is None and "blacklisted" in legacy:
archived = legacy["blacklisted"] archived = legacy["blacklisted"]
@@ -1052,6 +1297,9 @@ class EntryManager:
if tags is not None: if tags is not None:
entry["tags"] = tags entry["tags"] = tags
logger.debug(f"Updated tags for index {index}.") logger.debug(f"Updated tags for index {index}.")
if links is not None:
entry["links"] = self._normalize_links(links)
logger.debug(f"Updated links for index {index}.")
policy_updates: dict[str, Any] = {} policy_updates: dict[str, Any] = {}
if include_special_chars is not None: if include_special_chars is not None:
@@ -1075,7 +1323,12 @@ class EntryManager:
entry_policy.update(policy_updates) entry_policy.update(policy_updates)
entry["policy"] = entry_policy entry["policy"] = entry_policy
entry["modified_ts"] = int(time.time()) entry.setdefault(
"date_added",
self._iso_from_unix(int(entry.get("modified_ts", 0) or 0)),
)
entry["modified_ts"] = self._now_unix()
entry["date_modified"] = self._iso_from_unix(entry["modified_ts"])
data["entries"][str(index)] = entry data["entries"][str(index)] = entry
logger.debug( logger.debug(
@@ -1098,6 +1351,113 @@ class EntryManager:
) )
raise raise
def add_link(
self,
index: int,
target_id: int,
*,
relation: str = "related_to",
note: str = "",
) -> list[dict[str, Any]]:
"""Create or update a typed relationship between two entries."""
data = self._load_index()
entries = data.get("entries", {})
src = entries.get(str(index))
dst = entries.get(str(target_id))
if not isinstance(src, dict):
raise ValueError(f"Entry not found: {index}")
if not isinstance(dst, dict):
raise ValueError(f"Target entry not found: {target_id}")
if int(index) == int(target_id):
raise ValueError("Self-referential links are not allowed")
current_links = self._normalize_links(src.get("links", []))
relation_norm = str(relation or "related_to").strip().lower()
note_norm = str(note or "").strip()
candidate = {
"target_id": int(target_id),
"relation": relation_norm,
"note": note_norm,
}
current_links.append(candidate)
src["links"] = self._normalize_links(current_links)
src.setdefault(
"date_added",
self._iso_from_unix(int(src.get("modified_ts", 0) or self._now_unix())),
)
src["modified_ts"] = self._now_unix()
src["date_modified"] = self._iso_from_unix(src["modified_ts"])
self._save_index(data)
self.update_checksum()
self.backup_manager.create_backup()
return src["links"]
def remove_link(
self,
index: int,
target_id: int,
*,
relation: str | None = None,
) -> list[dict[str, Any]]:
"""Remove one or more links to ``target_id`` from an entry."""
data = self._load_index()
entries = data.get("entries", {})
src = entries.get(str(index))
if not isinstance(src, dict):
raise ValueError(f"Entry not found: {index}")
relation_norm = str(relation).strip().lower() if relation is not None else None
links = self._normalize_links(src.get("links", []))
filtered = []
for link in links:
matches_target = int(link.get("target_id", -1)) == int(target_id)
matches_relation = relation_norm is None or str(
link.get("relation", "")
).lower() == relation_norm
if matches_target and matches_relation:
continue
filtered.append(link)
if len(filtered) != len(links):
src["links"] = filtered
src.setdefault(
"date_added",
self._iso_from_unix(
int(src.get("modified_ts", 0) or self._now_unix())
),
)
src["modified_ts"] = self._now_unix()
src["date_modified"] = self._iso_from_unix(src["modified_ts"])
self._save_index(data)
self.update_checksum()
self.backup_manager.create_backup()
return src.get("links", [])
def get_links(self, index: int) -> list[dict[str, Any]]:
"""Return links for an entry with resolved target metadata."""
data = self._load_index()
entries = data.get("entries", {})
src = entries.get(str(index))
if not isinstance(src, dict):
raise ValueError(f"Entry not found: {index}")
links = self._normalize_links(src.get("links", []))
resolved: list[dict[str, Any]] = []
for link in links:
target_id = int(link.get("target_id"))
target = entries.get(str(target_id), {})
target_label = None
target_kind = None
if isinstance(target, dict):
target_label = target.get("label", target.get("website", ""))
target_kind = target.get("type", target.get("kind"))
resolved.append(
{
"target_id": target_id,
"relation": str(link.get("relation", "")),
"note": str(link.get("note", "")),
"target_label": target_label,
"target_kind": target_kind,
}
)
return resolved
def archive_entry(self, index: int) -> None: def archive_entry(self, index: int) -> None:
"""Mark the specified entry as archived.""" """Mark the specified entry as archived."""
self.modify_entry(index, archived=True) self.modify_entry(index, archived=True)
@@ -1147,7 +1507,7 @@ class EntryManager:
return entry.get("label", entry.get("website", "")).lower() return entry.get("label", entry.get("website", "")).lower()
if sort_by == "updated": if sort_by == "updated":
# sort newest first # sort newest first
return -int(entry.get("updated", 0)) return -int(entry.get("modified_ts", entry.get("updated", 0)) or 0)
raise ValueError("sort_by must be 'index', 'label', or 'updated'") raise ValueError("sort_by must be 'index', 'label', or 'updated'")
sorted_items = sorted(entries_data.items(), key=sort_key) sorted_items = sorted(entries_data.items(), key=sort_key)
@@ -1231,6 +1591,15 @@ class EntryManager:
"cyan", "cyan",
) )
) )
elif etype == EntryType.DOCUMENT.value:
print(colored(" Type: Document", "cyan"))
print(colored(f" Label: {entry.get('label', '')}", "cyan"))
print(
colored(
f" File Type: {entry.get('file_type', 'txt')}",
"cyan",
)
)
else: else:
print(colored(f" Label: {entry.get('label', '')}", "cyan")) print(colored(f" Label: {entry.get('label', '')}", "cyan"))
print( print(
@@ -1283,14 +1652,41 @@ class EntryManager:
) )
url = entry.get("url", "") if etype == EntryType.PASSWORD else None url = entry.get("url", "") if etype == EntryType.PASSWORD else None
tags = entry.get("tags", []) tags = entry.get("tags", [])
links = entry.get("links", [])
content = entry.get("content", "") if etype == EntryType.DOCUMENT else ""
file_type = (
entry.get("file_type", "") if etype == EntryType.DOCUMENT else ""
)
archived = entry.get("archived", entry.get("blacklisted", False)) archived = entry.get("archived", entry.get("blacklisted", False))
label_match = query_lower in label.lower() label_match = query_lower in label.lower()
username_match = bool(username) and query_lower in username.lower() username_match = bool(username) and query_lower in username.lower()
url_match = bool(url) and query_lower in url.lower() url_match = bool(url) and query_lower in url.lower()
tags_match = any(query_lower in str(t).lower() for t in tags) tags_match = any(query_lower in str(t).lower() for t in tags)
links_match = any(
query_lower in str(link.get("relation", "")).lower()
or query_lower in str(link.get("note", "")).lower()
or query_lower
in str(
(entries_data.get(str(link.get("target_id", ""))) or {}).get(
"label", ""
)
).lower()
for link in links
if isinstance(link, dict)
)
content_match = bool(content) and query_lower in str(content).lower()
file_type_match = bool(file_type) and query_lower in str(file_type).lower()
if label_match or username_match or url_match or tags_match: if (
label_match
or username_match
or url_match
or tags_match
or links_match
or content_match
or file_type_match
):
results.append( results.append(
( (
int(idx), int(idx),
@@ -1314,7 +1710,7 @@ class EntryManager:
data = self._load_index() data = self._load_index()
if "entries" in data and str(index) in data["entries"]: if "entries" in data and str(index) in data["entries"]:
entry = data["entries"][str(index)] entry = data["entries"][str(index)]
deleted_ts = int(time.time()) deleted_ts = self._now_unix()
entry_hash = hashlib.sha256( entry_hash = hashlib.sha256(
canonical_json_dumps(entry).encode("utf-8") canonical_json_dumps(entry).encode("utf-8")
).hexdigest() ).hexdigest()
+1
View File
@@ -15,6 +15,7 @@ class EntryType(str, Enum):
NOSTR = "nostr" NOSTR = "nostr"
KEY_VALUE = "key_value" KEY_VALUE = "key_value"
MANAGED_ACCOUNT = "managed_account" MANAGED_ACCOUNT = "managed_account"
DOCUMENT = "document"
# List of all entry type values for convenience # List of all entry type values for convenience
+290
View File
@@ -2228,6 +2228,81 @@ class PasswordManager:
print(colored(f"Error: Failed to add {operation_name}: {e}", "red")) print(colored(f"Error: Failed to add {operation_name}: {e}", "red"))
pause() pause()
def _terminal_document_editor(
self, initial_content: str = "", *, title: str = "Document Editor"
) -> str | None:
"""Simple line-oriented terminal editor.
Commands:
- `p`: print buffer
- `a`: append line
- `i <n>`: insert before line n
- `e <n>`: edit line n
- `d <n>`: delete line n
- `w`: save and exit
- `q`: cancel
"""
lines = initial_content.splitlines()
print(colored(f"\n{title}", "cyan"))
print(colored("Type 'h' for help.", "cyan"))
def print_help() -> None:
print(colored("Commands: p | a | i <n> | e <n> | d <n> | w | q", "cyan"))
def print_lines() -> None:
if not lines:
print(colored("[empty]", "yellow"))
return
for idx, line in enumerate(lines, start=1):
print(colored(f"{idx:>4}: {line}", "cyan"))
print_help()
print_lines()
while True:
raw = input("(doc-edit)> ").rstrip()
if not raw:
continue
cmd, *rest = raw.split(maxsplit=1)
cmd = cmd.lower()
if cmd == "h":
print_help()
continue
if cmd == "p":
print_lines()
continue
if cmd == "a":
lines.append(input("append> "))
continue
if cmd in {"i", "e", "d"}:
if not rest or not rest[0].strip().isdigit():
print(colored("Line number required.", "red"))
continue
line_no = int(rest[0].strip())
if line_no < 1 or line_no > max(len(lines), 1):
print(colored("Invalid line number.", "red"))
continue
if cmd == "i":
lines.insert(line_no - 1, input("insert> "))
elif cmd == "e":
if line_no > len(lines):
print(colored("Line does not exist.", "red"))
continue
lines[line_no - 1] = input("edit> ")
else:
if line_no > len(lines):
print(colored("Line does not exist.", "red"))
continue
lines.pop(line_no - 1)
continue
if cmd == "w":
return "\n".join(lines)
if cmd == "q":
return None
print(colored("Unknown command.", "red"))
@pause_logging_for_ui @pause_logging_for_ui
def handle_add_password(self) -> None: def handle_add_password(self) -> None:
self.entry_service.handle_add_password() self.entry_service.handle_add_password()
@@ -2591,6 +2666,69 @@ class PasswordManager:
print(color_text(f"Value: {value}", "deterministic")) print(color_text(f"Value: {value}", "deterministic"))
self._sync_and_pause() self._sync_and_pause()
@pause_logging_for_ui
def handle_add_document(self) -> None:
"""Add a document entry with editable text content."""
with self._entry_op_context("document"):
self._clear_header_add_entry("Document")
print("1. New Document")
print("2. Import Document File")
mode = input("Select option or press Enter to cancel: ").strip()
if not mode:
return
if mode not in {"1", "2"}:
print(colored("Invalid choice.", "red"))
return
if mode == "2":
file_path = input("Path to document file: ").strip()
if not file_path:
print(colored("Error: File path cannot be empty.", "red"))
return
label = input("Title override (optional): ").strip() or None
notes = input("Notes (optional): ").strip()
tags_input = input("Enter tags (comma-separated, optional): ").strip()
tags = (
[t.strip() for t in tags_input.split(",") if t.strip()]
if tags_input
else []
)
index = self.entry_manager.import_document_file(
file_path,
label=label,
notes=notes,
tags=tags,
)
else:
label = input("Title: ").strip()
if not label:
print(colored("Error: Title cannot be empty.", "red"))
return
file_type = (
input("File type/extension (default txt): ").strip() or "txt"
)
notes = input("Notes (optional): ").strip()
tags_input = input("Enter tags (comma-separated, optional): ").strip()
tags = (
[t.strip() for t in tags_input.split(",") if t.strip()]
if tags_input
else []
)
content = self._terminal_document_editor("", title=f"Editing: {label}")
if content is None:
print(colored("Document creation cancelled.", "yellow"))
return
index = self.entry_manager.add_document(
label,
content,
file_type=file_type,
notes=notes,
tags=tags,
)
self._mark_dirty()
print(colored(f"\n[+] Document entry added with ID {index}.\n", "green"))
self._sync_and_pause()
@pause_logging_for_ui @pause_logging_for_ui
def handle_add_managed_account(self) -> None: def handle_add_managed_account(self) -> None:
"""Add a managed account seed entry.""" """Add a managed account seed entry."""
@@ -2789,6 +2927,20 @@ class PasswordManager:
pause() pause()
refresh_entry() refresh_entry()
def export_document():
out = input(
"Output path (file or directory, blank for current directory): "
).strip()
try:
dest = self.entry_manager.export_document_file(
index, output_path=out or None
)
print(colored(f"Document exported to: {dest}", "green"))
except Exception as exc:
print(colored(f"Export failed: {exc}", "red"))
pause()
refresh_entry()
def get_options(): def get_options():
archived = current_entry.get( archived = current_entry.get(
"archived", current_entry.get("blacklisted", False) "archived", current_entry.get("blacklisted", False)
@@ -2813,6 +2965,8 @@ class PasswordManager:
EntryType.NOSTR.value, EntryType.NOSTR.value,
}: }:
options.append(("q", "Show QR codes", show_qr)) options.append(("q", "Show QR codes", show_qr))
if entry_type == EntryType.DOCUMENT.value:
options.append(("x", "Export Document to File", export_document))
return options return options
@@ -2854,6 +3008,30 @@ class PasswordManager:
self.last_update = time.time() self.last_update = time.time()
refresh_entry() refresh_entry()
def edit_document_content():
new_content = self._terminal_document_editor(
current_entry.get("content", ""),
title=f"Editing: {current_entry.get('label', 'Document')}",
)
if new_content is None:
print(colored("Edit cancelled.", "yellow"))
return
self.entry_manager.modify_entry(index, content=new_content)
self.is_dirty = True
self.last_update = time.time()
refresh_entry()
def edit_document_file_type():
current = current_entry.get("file_type", "txt")
new_file_type = (
input(f"New file type/extension (current: {current}): ").strip()
or current
)
self.entry_manager.modify_entry(index, file_type=new_file_type)
self.is_dirty = True
self.last_update = time.time()
refresh_entry()
def edit_username(): def edit_username():
new_username = input("New username: ").strip() new_username = input("New username: ").strip()
self.entry_manager.modify_entry(index, username=new_username) self.entry_manager.modify_entry(index, username=new_username)
@@ -2895,6 +3073,9 @@ class PasswordManager:
if entry_type == EntryType.KEY_VALUE.value: if entry_type == EntryType.KEY_VALUE.value:
options.append(("k", "Edit Key", edit_key)) options.append(("k", "Edit Key", edit_key))
options.append(("v", "Edit Value", edit_value)) options.append(("v", "Edit Value", edit_value))
elif entry_type == EntryType.DOCUMENT.value:
options.append(("c", "Edit Content", edit_document_content))
options.append(("f", "Edit File Type", edit_document_file_type))
if entry_type == EntryType.PASSWORD.value: if entry_type == EntryType.PASSWORD.value:
options.append(("u", "Edit Username", edit_username)) options.append(("u", "Edit Username", edit_username))
@@ -3228,6 +3409,81 @@ class PasswordManager:
label=new_label, label=new_label,
tags=tags, tags=tags,
) )
elif entry_type == EntryType.DOCUMENT.value:
label = entry.get("label", "")
content = entry.get("content", "")
file_type = entry.get("file_type", "txt")
blacklisted = entry.get("archived", False)
notes = entry.get("notes", "")
print(
colored(
f"Modifying document '{label}' (Index: {index}):",
"cyan",
)
)
new_label = (
input(f'Enter new title (leave blank to keep "{label}"): ').strip()
or label
)
new_file_type = (
input(
f'Enter file type/extension (leave blank to keep "{file_type}"): '
).strip()
or file_type
)
if input("Edit document content now? (y/N): ").strip().lower() == "y":
edited = self._terminal_document_editor(
content, title=f"Editing: {new_label}"
)
new_content = content if edited is None else edited
else:
new_content = content
blacklist_input = (
input(
f'Archive this entry? (Y/N, current: {"Y" if blacklisted else "N"}): '
)
.strip()
.lower()
)
if blacklist_input == "":
new_blacklisted = blacklisted
elif blacklist_input == "y":
new_blacklisted = True
elif blacklist_input == "n":
new_blacklisted = False
else:
self.notify(
"Invalid input for archived status. Keeping the current status.",
level="WARNING",
)
new_blacklisted = blacklisted
new_notes = (
input(
f'Enter new notes (leave blank to keep "{notes or "N/A"}"): '
).strip()
or notes
)
tags_input = input(
"Enter tags (comma-separated, leave blank to keep current): "
).strip()
tags = (
[t.strip() for t in tags_input.split(",") if t.strip()]
if tags_input
else None
)
self.entry_manager.modify_entry(
index,
archived=new_blacklisted,
notes=new_notes,
label=new_label,
content=new_content,
file_type=new_file_type,
tags=tags,
)
elif entry_type in ( elif entry_type in (
EntryType.KEY_VALUE.value, EntryType.KEY_VALUE.value,
EntryType.MANAGED_ACCOUNT.value, EntryType.MANAGED_ACCOUNT.value,
@@ -3632,6 +3888,36 @@ class PasswordManager:
print(color_text(f" Tags: {', '.join(tags)}", "index")) print(color_text(f" Tags: {', '.join(tags)}", "index"))
blacklisted = entry.get("archived", entry.get("blacklisted", False)) blacklisted = entry.get("archived", entry.get("blacklisted", False))
print(color_text(f" Archived: {'Yes' if blacklisted else 'No'}", "index")) print(color_text(f" Archived: {'Yes' if blacklisted else 'No'}", "index"))
elif etype == EntryType.DOCUMENT.value:
print(color_text(" Type: Document", "index"))
print(color_text(f" Title: {entry.get('label', '')}", "index"))
print(
color_text(
f" File Type: {entry.get('file_type', 'txt')}",
"index",
)
)
content = str(entry.get("content", ""))
preview = content if len(content) <= 240 else f"{content[:240]}..."
print(color_text(" Content Preview:", "index"))
print(color_text(f"{preview}", "index"))
notes = entry.get("notes", "")
if notes:
print(color_text(f" Notes: {notes}", "index"))
tags = entry.get("tags", [])
if tags:
print(color_text(f" Tags: {', '.join(tags)}", "index"))
if entry.get("date_added"):
print(color_text(f" Date Added: {entry.get('date_added')}", "index"))
if entry.get("date_modified"):
print(
color_text(
f" Date Modified: {entry.get('date_modified')}",
"index",
)
)
blacklisted = entry.get("archived", entry.get("blacklisted", False))
print(color_text(f" Archived: {'Yes' if blacklisted else 'No'}", "index"))
elif etype == EntryType.MANAGED_ACCOUNT.value: elif etype == EntryType.MANAGED_ACCOUNT.value:
print(color_text(" Type: Managed Account", "index")) print(color_text(" Type: Managed Account", "index"))
print(color_text(f" Label: {entry.get('label', '')}", "index")) print(color_text(f" Label: {entry.get('label', '')}", "index"))
@@ -3671,6 +3957,10 @@ class PasswordManager:
"index", "index",
) )
) )
if entry.get("date_added"):
print(color_text(f" Date Added: {entry.get('date_added')}", "index"))
if entry.get("date_modified"):
print(color_text(f" Date Modified: {entry.get('date_modified')}", "index"))
print("-" * 40) print("-" * 40)
@pause_logging_for_ui @pause_logging_for_ui
+3 -2
View File
@@ -137,7 +137,7 @@ def _prefer_entry(current: dict[str, Any], incoming: dict[str, Any]) -> bool:
def _union_fields_for_kind(kind: str) -> list[str]: def _union_fields_for_kind(kind: str) -> list[str]:
common = ["notes", "tags", "custom_fields"] common = ["notes", "tags", "custom_fields", "links"]
matrix = { matrix = {
"password": ["username", "url"], "password": ["username", "url"],
"totp": ["issuer"], "totp": ["issuer"],
@@ -147,6 +147,7 @@ def _union_fields_for_kind(kind: str) -> list[str]:
"pgp": ["public_key", "key_type"], "pgp": ["public_key", "key_type"],
"nostr": ["npub", "public_key"], "nostr": ["npub", "public_key"],
"seed": ["path", "network", "coin_type"], "seed": ["path", "network", "coin_type"],
"document": ["content", "file_type"],
} }
return matrix.get(kind, []) + common return matrix.get(kind, []) + common
@@ -170,7 +171,7 @@ def _merge_equal_ts_entries(
for field in union_fields: for field in union_fields:
pv = merged.get(field) pv = merged.get(field)
ov = other.get(field) ov = other.get(field)
if field in ("tags", "custom_fields"): if field in ("tags", "custom_fields", "links"):
merged[field] = _merge_list_union(pv, ov) merged[field] = _merge_list_union(pv, ov)
continue continue
if _is_empty(pv) and not _is_empty(ov): if _is_empty(pv) and not _is_empty(ov):
+142
View File
@@ -109,6 +109,148 @@ async def test_create_and_modify_ssh_entry(client):
assert calls["modify"][1]["notes"] == "x" assert calls["modify"][1]["notes"] == "x"
@pytest.mark.anyio
async def test_create_and_modify_document_entry(client):
cl, token = client
calls = {}
def add_document(label, content, **kwargs):
calls["create"] = (label, content, kwargs)
return 11
def modify(idx, **kwargs):
calls["modify"] = (idx, kwargs)
api.app.state.pm.entry_manager.add_document = add_document
api.app.state.pm.entry_manager.modify_entry = modify
headers = {"Authorization": f"Bearer {token}"}
res = await cl.post(
"/api/v1/entry",
json={
"type": "document",
"title": "Runbook",
"content": "line1\\nline2",
"file_type": "md",
"notes": "ops",
"tags": ["doc", "ops"],
},
headers=headers,
)
assert res.status_code == 200
assert res.json() == {"id": 11}
assert calls["create"] == (
"Runbook",
"line1\\nline2",
{"file_type": "md", "notes": "ops", "archived": False, "tags": ["doc", "ops"]},
)
res = await cl.put(
"/api/v1/entry/11",
json={"content": "line1\\nline2\\nline3", "file_type": "txt"},
headers=headers,
)
assert res.status_code == 200
assert calls["modify"][0] == 11
assert calls["modify"][1]["content"].endswith("line3")
assert calls["modify"][1]["file_type"] == "txt"
@pytest.mark.anyio
async def test_document_import_export_endpoints(client):
cl, token = client
calls = {}
def import_document(path, **kwargs):
calls["import"] = (path, kwargs)
return 33
def export_document(idx, **kwargs):
calls["export"] = (idx, kwargs)
return Path("/tmp/exported.md")
api.app.state.pm.entry_manager.import_document_file = import_document
api.app.state.pm.entry_manager.export_document_file = export_document
headers = {"Authorization": f"Bearer {token}"}
imported = await cl.post(
"/api/v1/entry/document/import",
json={"path": "/tmp/source.md", "label": "Imported", "notes": "n"},
headers=headers,
)
assert imported.status_code == 200
assert imported.json() == {"id": 33}
assert calls["import"] == (
"/tmp/source.md",
{"label": "Imported", "notes": "n", "archived": False, "tags": None},
)
exported = await cl.post(
"/api/v1/entry/33/document/export",
json={"path": "/tmp/out", "overwrite": True},
headers=headers,
)
assert exported.status_code == 200
assert exported.json() == {"path": "/tmp/exported.md"}
assert calls["export"] == (33, {"output_path": "/tmp/out", "overwrite": True})
@pytest.mark.anyio
async def test_entry_links_endpoints(client):
cl, token = client
calls = {}
def add_link(entry_id, target_id, **kwargs):
calls["add"] = (entry_id, target_id, kwargs)
return [{"target_id": target_id, "relation": kwargs["relation"], "note": kwargs["note"]}]
def remove_link(entry_id, target_id, **kwargs):
calls["remove"] = (entry_id, target_id, kwargs)
return []
def get_links(entry_id):
calls["get"] = entry_id
return [
{
"target_id": 2,
"relation": "references",
"note": "used by doc",
"target_label": "API Token",
"target_kind": "key_value",
}
]
api.app.state.pm.entry_manager.add_link = add_link
api.app.state.pm.entry_manager.remove_link = remove_link
api.app.state.pm.entry_manager.get_links = get_links
headers = {"Authorization": f"Bearer {token}"}
created = await cl.post(
"/api/v1/entry/1/links",
json={"target_id": 2, "relation": "references", "note": "used by doc"},
headers=headers,
)
assert created.status_code == 200
assert created.json()["links"][0]["target_id"] == 2
assert calls["add"] == (1, 2, {"relation": "references", "note": "used by doc"})
listed = await cl.get("/api/v1/entry/1/links", headers=headers)
assert listed.status_code == 200
assert listed.json()["entry_id"] == 1
assert listed.json()["links"][0]["target_kind"] == "key_value"
assert calls["get"] == 1
removed = await cl.request(
"DELETE",
"/api/v1/entry/1/links",
json={"target_id": 2, "relation": "references"},
headers=headers,
)
assert removed.status_code == 200
assert removed.json() == {"entry_id": 1, "links": []}
assert calls["remove"] == (1, 2, {"relation": "references"})
@pytest.mark.anyio @pytest.mark.anyio
async def test_update_entry_error(client): async def test_update_entry_error(client):
cl, token = client cl, token = client
+202
View File
@@ -586,6 +586,207 @@ def test_agent_token_issue_and_use_limit(monkeypatch, tmp_path):
assert second_payload["reason"] == "token_exhausted" assert second_payload["reason"] == "token_exhausted"
def test_agent_document_import_with_token_scope(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 import_document_file(self, file_path, **kwargs):
assert str(file_path).endswith("notes.md")
assert kwargs["label"] == "My Notes"
assert kwargs["notes"] == "imported"
assert kwargs["tags"] == ["work"]
assert kwargs["archived"] is False
return 12
def retrieve_entry(self, _i):
return {"kind": EntryType.DOCUMENT.value, "label": "My Notes", "file_type": "md"}
monkeypatch.setattr(agent_cli, "EntryService", DummyEntryService)
monkeypatch.setattr(
agent_cli,
"_load_policy",
lambda **kwargs: {
"version": 1,
"default_effect": "deny",
"rules": [
{
"id": "allow_doc_read",
"effect": "allow",
"operations": ["read"],
"kinds": ["document"],
"label_regex": ".*",
"path_regex": "^entry/.*$",
"fields": ["secret"],
}
],
"approvals": {"require_for": []},
"output": {"safe_output_default": True, "redact_fields": ["secret"]},
"export": {"allow_full_vault": True},
"secret_isolation": {"enabled": False, "high_risk_kinds": [], "unlock_ttl_sec": 300},
"allow_kinds": ["document"],
"deny_private_reveal": [],
"allow_export_import": True,
},
)
source = tmp_path / "notes.md"
source.write_text("# hello\n", encoding="utf-8")
issue = runner.invoke(
app,
[
"agent",
"token-issue",
"--name",
"doc-import",
"--scope",
"import",
"--kind",
"document",
"--uses",
"1",
"--ttl",
"600",
],
)
assert issue.exit_code == 0
token = json.loads(issue.stdout)["token"]
result = runner.invoke(
app,
[
"--fingerprint",
"ABC123",
"agent",
"document-import",
"--file",
str(source),
"--label",
"My Notes",
"--notes",
"imported",
"--tag",
"work",
"--token",
token,
],
)
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["status"] == "ok"
assert payload["index"] == 12
assert payload["kind"] == "document"
assert payload["file_type"] == "md"
assert payload["token_uses_remaining"] == 0
def test_agent_document_export_with_token_scope(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 retrieve_entry(self, _i):
return {"kind": EntryType.DOCUMENT.value, "label": "Spec"}
def export_document_file(self, _entry_id, out, **kwargs):
assert out == str(tmp_path)
assert kwargs["overwrite"] is True
return tmp_path / "Spec.md"
monkeypatch.setattr(agent_cli, "EntryService", DummyEntryService)
monkeypatch.setattr(
agent_cli,
"_load_policy",
lambda **kwargs: {
"version": 1,
"default_effect": "deny",
"rules": [
{
"id": "allow_doc_read",
"effect": "allow",
"operations": ["read"],
"kinds": ["document"],
"label_regex": ".*",
"path_regex": "^entry/.*$",
"fields": ["secret"],
}
],
"approvals": {"require_for": []},
"output": {"safe_output_default": True, "redact_fields": ["secret"]},
"export": {"allow_full_vault": True},
"secret_isolation": {"enabled": False, "high_risk_kinds": [], "unlock_ttl_sec": 300},
"allow_kinds": ["document"],
"deny_private_reveal": [],
"allow_export_import": True,
},
)
issue = runner.invoke(
app,
[
"agent",
"token-issue",
"--name",
"doc-export",
"--scope",
"export",
"--kind",
"document",
"--uses",
"1",
"--ttl",
"600",
],
)
assert issue.exit_code == 0
token = json.loads(issue.stdout)["token"]
result = runner.invoke(
app,
[
"--fingerprint",
"ABC123",
"agent",
"document-export",
"--entry-id",
"4",
"--out",
str(tmp_path),
"--overwrite",
"--token",
token,
],
)
assert result.exit_code == 0
payload = json.loads(result.stdout)
assert payload["status"] == "ok"
assert payload["index"] == 4
assert payload["kind"] == "document"
assert payload["output_path"] == str(tmp_path / "Spec.md")
assert payload["token_uses_remaining"] == 0
def test_agent_identity_create_list_revoke(monkeypatch, tmp_path): def test_agent_identity_create_list_revoke(monkeypatch, tmp_path):
import seedpass.core.agent_identity as identity_core import seedpass.core.agent_identity as identity_core
@@ -1444,6 +1645,7 @@ def test_agent_bootstrap_context(monkeypatch, tmp_path):
assert "agent identity-create" in payload["commands"]["identities"] assert "agent identity-create" in payload["commands"]["identities"]
assert "agent high-risk-unlock" in payload["commands"]["secret_isolation"] assert "agent high-risk-unlock" in payload["commands"]["secret_isolation"]
assert "agent export-check" in payload["commands"]["export_controls"] assert "agent export-check" in payload["commands"]["export_controls"]
assert "agent document-export <query>" in payload["commands"]["document_io"]
assert "agent posture-check" in payload["commands"]["posture"] assert "agent posture-check" in payload["commands"]["posture"]
assert "agent posture-remediate" in payload["commands"]["posture"] assert "agent posture-remediate" in payload["commands"]["posture"]
assert "identities" in payload assert "identities" in payload
+159
View File
@@ -1,5 +1,6 @@
import pytest import pytest
from types import SimpleNamespace from types import SimpleNamespace
from pathlib import Path
from typer.testing import CliRunner from typer.testing import CliRunner
from seedpass.cli import app from seedpass.cli import app
@@ -133,6 +134,14 @@ runner = CliRunner()
{"index": 7, "notes": "n"}, {"index": 7, "notes": "n"},
"8", "8",
), ),
(
"add-document",
"add_document",
["Doc", "--content", "hello", "--file-type", "md", "--notes", "n"],
("Doc", "hello"),
{"file_type": "md", "notes": "n", "tags": None, "archived": False},
"9",
),
], ],
) )
def test_entry_add_commands( def test_entry_add_commands(
@@ -161,3 +170,153 @@ def test_entry_add_commands(
assert called["args"] == expected_args assert called["args"] == expected_args
assert called["kwargs"] == expected_kwargs assert called["kwargs"] == expected_kwargs
assert called.get("sync") is True assert called.get("sync") is True
def test_entry_import_document_command(monkeypatch, tmp_path):
called = {}
def import_document_file(path, **kwargs):
called["args"] = (path,)
called["kwargs"] = kwargs
return 123
def start_background_vault_sync():
called["sync"] = True
pm = SimpleNamespace(
entry_manager=SimpleNamespace(import_document_file=import_document_file),
parent_seed="seed",
select_fingerprint=lambda fp: None,
start_background_vault_sync=start_background_vault_sync,
)
monkeypatch.setattr(cli_common, "PasswordManager", lambda: pm)
source = tmp_path / "doc.md"
source.write_text("hello", encoding="utf-8")
result = runner.invoke(
app,
[
"entry",
"import-document",
"--file",
str(source),
"--label",
"Doc",
"--notes",
"n",
],
)
assert result.exit_code == 0
assert "123" in result.stdout
assert called["args"] == (str(source),)
assert called["kwargs"] == {
"label": "Doc",
"notes": "n",
"tags": None,
"archived": False,
}
assert called.get("sync") is True
def test_entry_export_document_command(monkeypatch):
called = {}
def export_document_file(entry_id, output_path=None, **kwargs):
called["args"] = (entry_id, output_path)
called["kwargs"] = kwargs
return Path("/tmp/doc.txt")
pm = SimpleNamespace(
entry_manager=SimpleNamespace(export_document_file=export_document_file),
parent_seed="seed",
select_fingerprint=lambda fp: None,
start_background_vault_sync=lambda: None,
)
monkeypatch.setattr(cli_common, "PasswordManager", lambda: pm)
result = runner.invoke(
app,
[
"entry",
"export-document",
"--entry-id",
"9",
"--out",
"/tmp/outdir",
"--overwrite",
],
)
assert result.exit_code == 0
assert "/tmp/doc.txt" in result.stdout
assert called["args"] == (9, "/tmp/outdir")
assert called["kwargs"] == {"overwrite": True}
def test_entry_link_commands(monkeypatch):
called = {}
def add_link(entry_id, target_id, **kwargs):
called["add"] = (entry_id, target_id, kwargs)
return [{"target_id": target_id, "relation": kwargs["relation"], "note": kwargs["note"]}]
def remove_link(entry_id, target_id, **kwargs):
called["remove"] = (entry_id, target_id, kwargs)
return []
def get_links(entry_id):
called["get"] = entry_id
return [{"target_id": 3, "relation": "references", "note": "", "target_label": "X", "target_kind": "document"}]
def start_background_vault_sync():
called["sync"] = called.get("sync", 0) + 1
pm = SimpleNamespace(
entry_manager=SimpleNamespace(
add_link=add_link,
remove_link=remove_link,
get_links=get_links,
),
parent_seed="seed",
select_fingerprint=lambda fp: None,
start_background_vault_sync=start_background_vault_sync,
)
monkeypatch.setattr(cli_common, "PasswordManager", lambda: pm)
add_res = runner.invoke(
app,
[
"entry",
"link-add",
"--entry-id",
"1",
"--target-id",
"3",
"--relation",
"references",
"--note",
"note",
],
)
assert add_res.exit_code == 0
assert called["add"] == (1, 3, {"relation": "references", "note": "note"})
list_res = runner.invoke(app, ["entry", "links", "--entry-id", "1"])
assert list_res.exit_code == 0
assert '"target_id": 3' in list_res.stdout
assert called["get"] == 1
rm_res = runner.invoke(
app,
[
"entry",
"link-remove",
"--entry-id",
"1",
"--target-id",
"3",
"--relation",
"references",
],
)
assert rm_res.exit_code == 0
assert called["remove"] == (1, 3, {"relation": "references"})
+2 -1
View File
@@ -43,6 +43,7 @@ def _make_pm(called, locked=None):
handle_add_password=add, handle_add_password=add,
handle_add_totp=lambda: None, handle_add_totp=lambda: None,
handle_add_managed_account=lambda: None, handle_add_managed_account=lambda: None,
handle_add_document=lambda: None,
handle_retrieve_entry=retrieve, handle_retrieve_entry=retrieve,
handle_modify_entry=modify, handle_modify_entry=modify,
update_activity=update, update_activity=update,
@@ -81,7 +82,7 @@ def test_out_of_range_menu(monkeypatch, capsys):
def test_invalid_add_entry_submenu(monkeypatch, capsys): def test_invalid_add_entry_submenu(monkeypatch, capsys):
called = {"add": False, "retrieve": False, "modify": False} called = {"add": False, "retrieve": False, "modify": False}
pm, _ = _make_pm(called) pm, _ = _make_pm(called)
inputs = iter(["1", "9", "", ""]) inputs = iter(["1", "10", "", ""])
monkeypatch.setattr(main, "timed_input", lambda *_: next(inputs)) monkeypatch.setattr(main, "timed_input", lambda *_: next(inputs))
monkeypatch.setattr("builtins.input", lambda *_: next(inputs)) monkeypatch.setattr("builtins.input", lambda *_: next(inputs))
with pytest.raises(SystemExit): with pytest.raises(SystemExit):
+6
View File
@@ -371,6 +371,12 @@ class TestEntryService:
digits=None, digits=None,
key=None, key=None,
value=None, value=None,
content=None,
file_type=None,
custom_fields=None,
tags=None,
links=None,
archived=None,
) )
mock_manager.start_background_vault_sync.assert_called_once() mock_manager.start_background_vault_sync.assert_called_once()
+45
View File
@@ -0,0 +1,45 @@
from pathlib import Path
from tempfile import TemporaryDirectory
from helpers import TEST_PASSWORD, TEST_SEED, create_vault
from seedpass.core.backup import BackupManager
from seedpass.core.config_manager import ConfigManager
from seedpass.core.entry_management import EntryManager
def _make_entry_manager(tmp_path: Path) -> EntryManager:
vault, _ = create_vault(tmp_path, TEST_SEED, TEST_PASSWORD)
cfg_mgr = ConfigManager(vault, tmp_path)
backup_mgr = BackupManager(tmp_path, cfg_mgr)
return EntryManager(vault, backup_mgr)
def test_import_document_file_roundtrip():
with TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
entry_mgr = _make_entry_manager(base)
source = base / "notes.md"
source.write_text("# hello\nworld\n", encoding="utf-8")
idx = entry_mgr.import_document_file(source, notes="imported", tags=["docs"])
entry = entry_mgr.retrieve_entry(idx)
assert entry is not None
assert entry["kind"] == "document"
assert entry["label"] == "notes"
assert entry["file_type"] == "md"
assert entry["content"] == "# hello\nworld\n"
assert entry["notes"] == "imported"
assert entry["tags"] == ["docs"]
def test_export_document_file_writes_expected_content():
with TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
entry_mgr = _make_entry_manager(base)
idx = entry_mgr.add_document("Runbook", "line1\nline2", file_type="txt")
out_dir = base / "exports"
dest = entry_mgr.export_document_file(idx, output_path=out_dir)
assert dest.exists()
assert dest.suffix == ".txt"
assert dest.read_text(encoding="utf-8") == "line1\nline2"
+10 -1
View File
@@ -29,7 +29,7 @@ def test_add_and_retrieve_entry():
index = entry_mgr.add_entry("example.com", 12, "user", custom_fields=custom) index = entry_mgr.add_entry("example.com", 12, "user", custom_fields=custom)
entry = entry_mgr.retrieve_entry(index) entry = entry_mgr.retrieve_entry(index)
assert entry == { expected = {
"label": "example.com", "label": "example.com",
"length": 12, "length": 12,
"username": "user", "username": "user",
@@ -41,6 +41,9 @@ def test_add_and_retrieve_entry():
"custom_fields": custom, "custom_fields": custom,
"tags": [], "tags": [],
} }
assert entry.items() >= expected.items()
assert "date_added" in entry
assert "date_modified" in entry
data = enc_mgr.load_json_data(entry_mgr.index_file) data = enc_mgr.load_json_data(entry_mgr.index_file)
assert str(index) in data.get("entries", {}) assert str(index) in data.get("entries", {})
@@ -58,6 +61,7 @@ def test_add_and_retrieve_entry():
("add_seed", "seed"), ("add_seed", "seed"),
("add_key_value", "key_value"), ("add_key_value", "key_value"),
("add_managed_account", "managed_account"), ("add_managed_account", "managed_account"),
("add_document", "document"),
], ],
) )
def test_round_trip_entry_types(method, expected_type): def test_round_trip_entry_types(method, expected_type):
@@ -74,6 +78,8 @@ def test_round_trip_entry_types(method, expected_type):
index = 0 index = 0
elif method == "add_key_value": elif method == "add_key_value":
index = entry_mgr.add_key_value("label", "k1", "val") index = entry_mgr.add_key_value("label", "k1", "val")
elif method == "add_document":
index = entry_mgr.add_document("doc", "hello", file_type="txt")
else: else:
if method == "add_ssh_key": if method == "add_ssh_key":
index = entry_mgr.add_ssh_key("ssh", TEST_SEED) index = entry_mgr.add_ssh_key("ssh", TEST_SEED)
@@ -87,6 +93,8 @@ def test_round_trip_entry_types(method, expected_type):
entry = entry_mgr.retrieve_entry(index) entry = entry_mgr.retrieve_entry(index)
assert entry["type"] == expected_type assert entry["type"] == expected_type
assert entry["kind"] == expected_type assert entry["kind"] == expected_type
assert "date_added" in entry
assert "date_modified" in entry
data = enc_mgr.load_json_data(entry_mgr.index_file) data = enc_mgr.load_json_data(entry_mgr.index_file)
assert data["entries"][str(index)]["type"] == expected_type assert data["entries"][str(index)]["type"] == expected_type
assert data["entries"][str(index)]["kind"] == expected_type assert data["entries"][str(index)]["kind"] == expected_type
@@ -120,6 +128,7 @@ def test_legacy_entry_defaults_to_password():
("add_seed", ("seed", TEST_SEED)), ("add_seed", ("seed", TEST_SEED)),
("add_key_value", ("label", "k1", "val")), ("add_key_value", ("label", "k1", "val")),
("add_managed_account", ("acct", TEST_SEED)), ("add_managed_account", ("acct", TEST_SEED)),
("add_document", ("doc", "body")),
], ],
) )
def test_add_default_archived_false(method, args): def test_add_default_archived_false(method, args):
+63
View File
@@ -0,0 +1,63 @@
import sys
from pathlib import Path
from tempfile import TemporaryDirectory
from helpers import create_vault, TEST_PASSWORD, TEST_SEED
sys.path.append(str(Path(__file__).resolve().parents[1]))
from seedpass.core.backup import BackupManager
from seedpass.core.config_manager import ConfigManager
from seedpass.core.entry_management import EntryManager
def _entry_manager(tmp_path: Path) -> EntryManager:
vault, _ = create_vault(tmp_path, TEST_SEED, TEST_PASSWORD)
cfg_mgr = ConfigManager(vault, tmp_path)
backup_mgr = BackupManager(tmp_path, cfg_mgr)
return EntryManager(vault, backup_mgr)
def test_links_add_list_remove_and_search():
with TemporaryDirectory() as tmpdir:
em = _entry_manager(Path(tmpdir))
src = em.add_document("Project Plan", "v1", file_type="md")
dst = em.add_key_value("API Token", "token", "abc")
links = em.add_link(src, dst, relation="references", note="used by deploy")
assert links == [
{"target_id": dst, "relation": "references", "note": "used by deploy"}
]
resolved = em.get_links(src)
assert resolved[0]["target_id"] == dst
assert resolved[0]["relation"] == "references"
assert resolved[0]["target_label"] == "API Token"
assert resolved[0]["target_kind"] == "key_value"
# Query by relation, note, and target label should match source.
assert em.search_entries("references")[0][0] == src
assert em.search_entries("deploy")[0][0] == src
assert em.search_entries("api token")[0][0] == src
remaining = em.remove_link(src, dst, relation="references")
assert remaining == []
assert em.get_links(src) == []
def test_links_reject_invalid_targets_and_self_link():
with TemporaryDirectory() as tmpdir:
em = _entry_manager(Path(tmpdir))
idx = em.add_entry("example", 12)
try:
em.add_link(idx, idx)
assert False, "Expected self-link validation failure"
except ValueError as exc:
assert "Self-referential" in str(exc)
try:
em.add_link(idx, 9999)
assert False, "Expected missing target validation failure"
except ValueError as exc:
assert "Target entry not found" in str(exc)
+33
View File
@@ -68,6 +68,14 @@ def _add_entry_for_type(entry_mgr: EntryManager, entry_type: EntryType) -> int:
return entry_mgr.add_managed_account( return entry_mgr.add_managed_account(
"Managed Entry", TEST_SEED, notes="managed-note", tags=["managed"] "Managed Entry", TEST_SEED, notes="managed-note", tags=["managed"]
) )
if entry_type == EntryType.DOCUMENT:
return entry_mgr.add_document(
"Document Entry",
"alpha\nbeta\ngamma",
file_type="md",
notes="doc-note",
tags=["doc"],
)
raise AssertionError(f"Unhandled entry type in test: {entry_type}") raise AssertionError(f"Unhandled entry type in test: {entry_type}")
@@ -86,6 +94,8 @@ def test_each_entry_type_roundtrip_retrieve_and_filter(tmp_path, entry_type: Ent
assert entry["kind"] == entry_type.value assert entry["kind"] == entry_type.value
assert entry["label"] assert entry["label"]
assert "modified_ts" not in entry assert "modified_ts" not in entry
assert "date_added" in entry
assert "date_modified" in entry
entries = entry_mgr.list_entries( entries = entry_mgr.list_entries(
filter_kinds=[entry_type.value], include_archived=True, verbose=False filter_kinds=[entry_type.value], include_archived=True, verbose=False
@@ -126,3 +136,26 @@ def test_each_entry_type_roundtrip_retrieve_and_filter(tmp_path, entry_type: Ent
if entry_type == EntryType.MANAGED_ACCOUNT: if entry_type == EntryType.MANAGED_ACCOUNT:
assert entry["word_count"] == 12 assert entry["word_count"] == 12
assert entry["fingerprint"] assert entry["fingerprint"]
if entry_type == EntryType.DOCUMENT:
assert entry["file_type"] == "md"
assert "alpha" in entry["content"]
@pytest.mark.parametrize("entry_type", list(EntryType))
def test_each_entry_type_sets_and_updates_dates(
tmp_path, monkeypatch, entry_type: EntryType
):
entry_mgr = _make_entry_manager(tmp_path)
monkeypatch.setattr(entry_mgr, "_now_unix", lambda: 100)
index = _add_entry_for_type(entry_mgr, entry_type)
created = entry_mgr.retrieve_entry(index)
assert created is not None
assert created["date_added"] == entry_mgr._iso_from_unix(100)
assert created["date_modified"] == entry_mgr._iso_from_unix(100)
monkeypatch.setattr(entry_mgr, "_now_unix", lambda: 200)
entry_mgr.modify_entry(index, label=f"{created['label']} updated")
updated = entry_mgr.retrieve_entry(index)
assert updated is not None
assert updated["date_added"] == entry_mgr._iso_from_unix(100)
assert updated["date_modified"] == entry_mgr._iso_from_unix(200)
+32
View File
@@ -64,3 +64,35 @@ def test_full_sync_roundtrip(dummy_nostr_client):
pm_b.entry_manager.clear_cache() pm_b.entry_manager.clear_cache()
labels = [e[1] for e in pm_b.entry_manager.list_entries()] labels = [e[1] for e in pm_b.entry_manager.list_entries()]
assert sorted(labels) == ["site1", "site2"] assert sorted(labels) == ["site1", "site2"]
def test_large_document_sync_roundtrip(dummy_nostr_client):
client, relay = dummy_nostr_client
with TemporaryDirectory() as tmpdir:
base = Path(tmpdir)
dir_a = base / "A"
dir_b = base / "B"
dir_a.mkdir()
dir_b.mkdir()
pm_a = _init_pm(dir_a, client)
pm_b = _init_pm(dir_b, client)
large_content = "line\n" * 120_000
doc_id = pm_a.entry_manager.add_document(
"LargeDoc",
large_content,
file_type="txt",
notes="large-payload",
tags=["sync", "large"],
)
pm_a.sync_vault()
result = pm_b.attempt_initial_sync()
assert result is True
restored = pm_b.entry_manager.retrieve_entry(doc_id)
assert restored is not None
assert restored["kind"] == "document"
assert restored["file_type"] == "txt"
assert restored["content"] == large_content
+11 -11
View File
@@ -25,17 +25,17 @@ def test_add_and_modify_key_value():
idx = em.add_key_value("API entry", "api_key", "abc123", notes="token") idx = em.add_key_value("API entry", "api_key", "abc123", notes="token")
entry = em.retrieve_entry(idx) entry = em.retrieve_entry(idx)
assert entry == { assert entry["type"] == "key_value"
"type": "key_value", assert entry["kind"] == "key_value"
"kind": "key_value", assert entry["label"] == "API entry"
"label": "API entry", assert entry["key"] == "api_key"
"key": "api_key", assert entry["value"] == "abc123"
"value": "abc123", assert entry["notes"] == "token"
"notes": "token", assert entry["archived"] is False
"archived": False, assert entry["custom_fields"] == []
"custom_fields": [], assert entry["tags"] == []
"tags": [], assert "date_added" in entry
} assert "date_modified" in entry
# Appears in listing # Appears in listing
assert em.list_entries() == [(idx, "API entry", None, None, False)] assert em.list_entries() == [(idx, "API entry", None, None, False)]
@@ -49,9 +49,11 @@ def test_migrate_iterations(tmp_path, monkeypatch, iterations):
calls: list[int] = [] calls: list[int] = []
orig_derive = enc_module._derive_legacy_key_from_password orig_derive = enc_module._derive_legacy_key_from_password
def tracking_derive(password: str, iterations: int = 100_000) -> bytes: def tracking_derive(
password: str, iterations: int = 100_000, salt: bytes = b""
) -> bytes:
calls.append(iterations) calls.append(iterations)
return orig_derive(password, iterations=iterations) return orig_derive(password, iterations=iterations, salt=salt)
monkeypatch.setattr(enc_module, "_derive_legacy_key_from_password", tracking_derive) monkeypatch.setattr(enc_module, "_derive_legacy_key_from_password", tracking_derive)
@@ -60,7 +62,11 @@ def test_migrate_iterations(tmp_path, monkeypatch, iterations):
vault.load_index() vault.load_index()
assert prompts == [1] assert prompts == [1]
expected = [50_000] if iterations == 50_000 else [50_000, 100_000] expected = (
[50_000, 100_000, 50_000]
if iterations == 50_000
else [50_000, 100_000, 50_000, 100_000]
)
assert calls == expected assert calls == expected
cfg = ConfigManager(vault, tmp_path) cfg = ConfigManager(vault, tmp_path)
+2 -2
View File
@@ -40,8 +40,8 @@ def test_sort_by_updated():
idx1 = em.add_entry("beta.com", 8, "u1") idx1 = em.add_entry("beta.com", 8, "u1")
data = em._load_index(force_reload=True) data = em._load_index(force_reload=True)
data["entries"][str(idx0)]["updated"] = 1 data["entries"][str(idx0)]["modified_ts"] = 1
data["entries"][str(idx1)]["updated"] = 2 data["entries"][str(idx1)]["modified_ts"] = 2
em._save_index(data) em._save_index(data)
result = em.list_entries(sort_by="updated") result = em.list_entries(sort_by="updated")
+24 -24
View File
@@ -74,18 +74,18 @@ def test_handle_add_password(monkeypatch, dummy_nostr_client, capsys):
assert entries == [(0, "Example", "", "", False)] assert entries == [(0, "Example", "", "", False)]
entry = entry_mgr.retrieve_entry(0) entry = entry_mgr.retrieve_entry(0)
assert entry == { assert entry["label"] == "Example"
"label": "Example", assert entry["length"] == DEFAULT_PASSWORD_LENGTH
"length": DEFAULT_PASSWORD_LENGTH, assert entry["username"] == ""
"username": "", assert entry["url"] == ""
"url": "", assert entry["archived"] is False
"archived": False, assert entry["type"] == "password"
"type": "password", assert entry["kind"] == "password"
"kind": "password", assert entry["notes"] == ""
"notes": "", assert entry["custom_fields"] == []
"custom_fields": [], assert entry["tags"] == []
"tags": [], assert "date_added" in entry
} assert "date_modified" in entry
assert f"pw-0-{DEFAULT_PASSWORD_LENGTH}" in out assert f"pw-0-{DEFAULT_PASSWORD_LENGTH}" in out
@@ -194,18 +194,18 @@ def test_handle_add_password_quick_mode(monkeypatch, dummy_nostr_client, capsys)
assert entries == [(0, "Example", "", "", False)] assert entries == [(0, "Example", "", "", False)]
entry = entry_mgr.retrieve_entry(0) entry = entry_mgr.retrieve_entry(0)
assert entry == { assert entry["label"] == "Example"
"label": "Example", assert entry["length"] == DEFAULT_PASSWORD_LENGTH
"length": DEFAULT_PASSWORD_LENGTH, assert entry["username"] == ""
"username": "", assert entry["url"] == ""
"url": "", assert entry["archived"] is False
"archived": False, assert entry["type"] == "password"
"type": "password", assert entry["kind"] == "password"
"kind": "password", assert entry["notes"] == ""
"notes": "", assert entry["custom_fields"] == []
"custom_fields": [], assert entry["tags"] == []
"tags": [], assert "date_added" in entry
} assert "date_modified" in entry
assert f"pw-0-{DEFAULT_PASSWORD_LENGTH}" in out assert f"pw-0-{DEFAULT_PASSWORD_LENGTH}" in out
@@ -74,6 +74,12 @@ def entry_manager():
{"url": "seed_url"}, {"url": "seed_url"},
"does not support fields: url", "does not support fields: url",
), ),
(
EntryType.DOCUMENT,
lambda em: em.add_document("Doc", "body"),
{"username": "doc_user"},
"does not support fields: username",
),
], ],
) )
def test_modify_entry_invalid_fields( def test_modify_entry_invalid_fields(
+9 -9
View File
@@ -24,15 +24,15 @@ def test_nostr_key_determinism():
idx = entry_mgr.add_nostr_key("main", TEST_SEED) idx = entry_mgr.add_nostr_key("main", TEST_SEED)
entry = entry_mgr.retrieve_entry(idx) entry = entry_mgr.retrieve_entry(idx)
assert entry == { assert entry["type"] == "nostr"
"type": "nostr", assert entry["kind"] == "nostr"
"kind": "nostr", assert entry["index"] == idx
"index": idx, assert entry["label"] == "main"
"label": "main", assert entry["notes"] == ""
"notes": "", assert entry["archived"] is False
"archived": False, assert entry["tags"] == []
"tags": [], assert "date_added" in entry
} assert "date_modified" in entry
npub1, nsec1 = entry_mgr.get_nostr_key_pair(idx, TEST_SEED) npub1, nsec1 = entry_mgr.get_nostr_key_pair(idx, TEST_SEED)
npub2, nsec2 = entry_mgr.get_nostr_key_pair(idx, TEST_SEED) npub2, nsec2 = entry_mgr.get_nostr_key_pair(idx, TEST_SEED)
@@ -12,10 +12,12 @@ import seedpass.core.encryption as enc_module
def test_legacy_password_only_fallback(monkeypatch, tmp_path, caplog): def test_legacy_password_only_fallback(monkeypatch, tmp_path, caplog):
calls: list[int] = [] calls: list[int] = []
def _fast_legacy_key(password: str, iterations: int = 100_000) -> bytes: def _fast_legacy_key(
password: str, iterations: int = 100_000, salt: bytes = b""
) -> bytes:
calls.append(iterations) calls.append(iterations)
normalized = unicodedata.normalize("NFKD", password).strip().encode("utf-8") normalized = unicodedata.normalize("NFKD", password).strip().encode("utf-8")
key = hashlib.pbkdf2_hmac("sha256", normalized, b"", 1, dklen=32) key = hashlib.pbkdf2_hmac("sha256", normalized, salt, 1, dklen=32)
return base64.urlsafe_b64encode(key) return base64.urlsafe_b64encode(key)
# Speed up legacy key derivation # Speed up legacy key derivation
@@ -35,4 +37,4 @@ def test_legacy_password_only_fallback(monkeypatch, tmp_path, caplog):
assert enc_mgr.decrypt_and_save_index_from_nostr(encrypted) assert enc_mgr.decrypt_and_save_index_from_nostr(encrypted)
assert vault.load_index() == data assert vault.load_index() == data
assert any("legacy password-only" in rec.message for rec in caplog.records) assert any("legacy password-only" in rec.message for rec in caplog.records)
assert calls == [50_000, 50_000] assert calls == [50_000, 50_000, 100_000, 50_000]
+20 -20
View File
@@ -34,27 +34,27 @@ def test_seed_phrase_determinism():
entry12 = entry_mgr.retrieve_entry(idx_12) entry12 = entry_mgr.retrieve_entry(idx_12)
entry24 = entry_mgr.retrieve_entry(idx_24) entry24 = entry_mgr.retrieve_entry(idx_24)
assert entry12 == { assert entry12["type"] == "seed"
"type": "seed", assert entry12["kind"] == "seed"
"kind": "seed", assert entry12["index"] == idx_12
"index": idx_12, assert entry12["label"] == "seed12"
"label": "seed12", assert entry12["word_count"] == 12
"word_count": 12, assert entry12["notes"] == ""
"notes": "", assert entry12["archived"] is False
"archived": False, assert entry12["tags"] == []
"tags": [], assert "date_added" in entry12
} assert "date_modified" in entry12
assert entry24 == { assert entry24["type"] == "seed"
"type": "seed", assert entry24["kind"] == "seed"
"kind": "seed", assert entry24["index"] == idx_24
"index": idx_24, assert entry24["label"] == "seed24"
"label": "seed24", assert entry24["word_count"] == 24
"word_count": 24, assert entry24["notes"] == ""
"notes": "", assert entry24["archived"] is False
"archived": False, assert entry24["tags"] == []
"tags": [], assert "date_added" in entry24
} assert "date_modified" in entry24
assert phrase12_a not in entry12.values() assert phrase12_a not in entry12.values()
assert phrase24_a not in entry24.values() assert phrase24_a not in entry24.values()
+10 -9
View File
@@ -22,15 +22,16 @@ def test_add_and_retrieve_ssh_key_pair():
index = entry_mgr.add_ssh_key("ssh", TEST_SEED) index = entry_mgr.add_ssh_key("ssh", TEST_SEED)
entry = entry_mgr.retrieve_entry(index) entry = entry_mgr.retrieve_entry(index)
assert entry == { assert entry["type"] == "ssh"
"type": "ssh", assert entry["kind"] == "ssh"
"kind": "ssh", assert entry["index"] == index
"index": index, assert entry["label"] == "ssh"
"label": "ssh", assert entry["notes"] == ""
"notes": "", assert entry["archived"] is False
"archived": False, assert entry["tags"] == []
"tags": [], assert entry["custom_fields"] == []
} assert "date_added" in entry
assert "date_modified" in entry
priv1, pub1 = entry_mgr.get_ssh_key_pair(index, TEST_SEED) priv1, pub1 = entry_mgr.get_ssh_key_pair(index, TEST_SEED)
priv2, pub2 = entry_mgr.get_ssh_key_pair(index, TEST_SEED) priv2, pub2 = entry_mgr.get_ssh_key_pair(index, TEST_SEED)