diff --git a/CHANGELOG.md b/CHANGELOG.md index 81e5c52..795a0de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,15 @@ Format based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ### Added +- **Manual edits made in Bitwarden are no longer silently reverted on re-run** (issue #30). kp2bw is + KeePass-authoritative, so any difference on an existing item used to be overwritten with the KeePass value -- quietly + undoing a title you fixed, a note you added or a URI you corrected in Bitwarden. Every item kp2bw writes now carries a + hidden-from-noise `KP2BW_SYNC` content signature; a re-run that finds an item's current content no longer matching + that stamp knows a *user* edited it (kp2bw's own writes restamp, so they never self-trip) and **preserves the edit**, + reporting it as `protected` in the summary instead of clobbering it. `--force-update` (env `KP2BW_FORCE_UPDATE`) makes + KeePass win regardless. Unchanged re-runs stay idempotent (the stamp is excluded from the content diff), legacy/ + unstamped items keep current behaviour, and `--strip-ids` now removes `KP2BW_SYNC` alongside `KP2BW_ID`. + - **`--report-uris keepass|bitwarden` -- a read-only URI collision report** (env `KP2BW_REPORT_URIS`). Groups every login URL by registrable domain (a curated two-level public-suffix heuristic, so `10bis.co.il` stays whole) and lists the domains with more than one host -- exactly the logins that all surface together under Bitwarden's base-domain diff --git a/README.md b/README.md index 5932ff6..5e555a6 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ with advantages over the built-in Bitwarden importer: -
Passkey migration KeePassXC FIDO2/passkey credentials (KPEX_PASSKEY_*) are converted to Bitwarden fido2Credentials.
-
Custom properties & attachments Imported as Bitwarden custom fields or attachments (values >10k chars auto-upload as files).
-
Long notes handling Notes exceeding 10k chars are uploaded as notes.txt attachments.
--
Idempotent re-runs that sync changes Safe to run repeatedly; existing entries are updated in place when their KeePass content changed (notes, credentials, URIs, fields) and never duplicated. Each item is stamped with its KeePass UUID in a KP2BW_ID field, so distinct entries that share a title stay separate and a re-run is matched by identity rather than title.
Disable updates with --no-update.
+-
Idempotent re-runs that sync changes Safe to run repeatedly; existing entries are updated in place when their KeePass content changed (notes, credentials, URIs, fields) and never duplicated. Each item is stamped with its KeePass UUID in a KP2BW_ID field, so distinct entries that share a title stay separate and a re-run is matched by identity rather than title. Edits you make in Bitwarden are protected: a re-run preserves them instead of reverting to KeePass (a KP2BW_SYNC content stamp tells your edit apart from kp2bw's own writes); --force-update overrides.
Disable updates with --no-update.
-
Nested folders KeePass folder hierarchy is recreated in Bitwarden.
-
Recycle Bin filtering Deleted entries are automatically excluded.
-
Expiry awareness Expired entries are marked [EXPIRED] in notes; optionally skip them entirely with --skip-expired.
@@ -84,7 +84,7 @@ kp2bw [-h] [-V] [-k PASSWORD] [-K FILE] [-b PASSWORD] [-o ID] [-t TAG [TAG ...]] [-c ID] [--path-to-name | --no-path-to-name] [--path-to-name-skip N] [--skip-expired | --no-skip-expired] [--include-recycle-bin | --no-include-recycle-bin] - [--metadata | --no-metadata] [--update | --no-update] + [--metadata | --no-metadata] [--update | --no-update] [--force-update] [--include-oversize-secrets] [--uri-match MODE] [--interpret-uri-syntax | --no-interpret-uri-syntax] [--migrate-uris] [--report-uris SOURCE] [--strip-ids] [-y] [-v] [-d] @@ -106,12 +106,13 @@ kp2bw [-h] [-V] [-k PASSWORD] [-K FILE] [-b PASSWORD] [-o ID] | `--include-recycle-bin` | Include Recycle Bin entries (excluded by default) | `KP2BW_INCLUDE_RECYCLE_BIN` | | `--metadata` / `--no-metadata` | Toggle KeePass tags/expiry as a `KP2BW_META` field (default: on) | `KP2BW_MIGRATE_METADATA` | | `--update` / `--no-update` | Update existing entries changed in KeePass (default: on) | `KP2BW_UPDATE` | +| `--force-update` | Overwrite items even if edited in Bitwarden since the last run (default: protect such edits) | `KP2BW_FORCE_UPDATE` | | `--include-oversize-secrets` | Offload over-limit secret fields[^offload] to a `.txt` attachment instead of dropping them (default: off) | `KP2BW_INCLUDE_OVERSIZE_SECRETS` | | `--uri-match MODE` | Match mode for plain URLs: `default`(default, account default)/`domain`/`host`/`startswith`/`exact`/`regex`/`never` | `KP2BW_URI_MATCH` | | `--interpret-uri-syntax` | Honor KeePassXC quote/wildcard URL syntax on additional URLs (default: on; `--no-…` for literal) | `KP2BW_INTERPRET_URI_SYNTAX` | | `--migrate-uris` | Upgrade existing items: re-fold legacy `KP2A_URL*`/`AndroidApp` fields into login URIs, then exit (no KeePass) | `KP2BW_MIGRATE_URIS` | | `--report-uris SOURCE` | Print a read-only URI collision report (`keepass` or `bitwarden`) and exit; lists registrable domains with multiple hosts | `KP2BW_REPORT_URIS` | -| `--strip-ids` | Finalize: remove the `KP2BW_ID` dedup stamp from migrated items, then exit (no migration; no KeePass db) | `KP2BW_STRIP_IDS` | +| `--strip-ids` | Finalize: remove the `KP2BW_ID`/`KP2BW_SYNC` stamps from migrated items, then exit (no migration; no KeePass db) | `KP2BW_STRIP_IDS` | | `-y, --yes` | Skip the Bitwarden CLI setup confirmation prompt | `KP2BW_YES` | | `-v, --verbose` | Verbose output | `KP2BW_VERBOSE` | | `-d, --debug` | Debug output — includes third-party library logs | `KP2BW_DEBUG` | @@ -142,6 +143,13 @@ Too many subdomains autofilling together? Under base-domain matching, every logi registrable domains with multiple hosts — so you can see which entries pile up and switch those to **Host** match (or flip your Bitwarden account's default URI match detection to Host). It changes nothing; it just lists. +Because kp2bw is KeePass-authoritative, a re-run would normally overwrite any difference on an existing item — quietly +undoing a title, note or URI you fixed in Bitwarden. It doesn't: every item kp2bw writes carries a `KP2BW_SYNC` content +signature, and a re-run that finds an item's current content no longer matching that stamp knows *you* edited it +(kp2bw's own writes restamp, so they never self-trip) and **preserves your edit**, listing it as `protected` in the +summary. Pass `--force-update` (env `KP2BW_FORCE_UPDATE`) to make KeePass win regardless. Unchanged re-runs stay +idempotent, and items imported before this existed are adopted normally on their first re-run. + Every migrated item carries a plain-text `KP2BW_ID` custom field — the KeePass UUID kp2bw uses to match entries on re-runs so nothing duplicates. Once you're satisfied the migration is complete and you're ready to fully adopt Bitwarden, run diff --git a/src/kp2bw/AGENTS.md b/src/kp2bw/AGENTS.md index 10da054..9ff0740 100644 --- a/src/kp2bw/AGENTS.md +++ b/src/kp2bw/AGENTS.md @@ -78,13 +78,26 @@ src/kp2bw/ failing that it creates a new item. This stops distinct same-titled entries from collapsing onto one item (data loss) and keeps re-runs idempotent across title/folder edits. The legacy adoption is a one-time path for vaults imported before stable identity. -- `--strip-ids` (`KP2BW_STRIP_IDS`, default off) is the finalize-mode inverse of the stamp above: it short-circuits +- Manual-edit protection (issue #30, `--force-update` / `KP2BW_FORCE_UPDATE`, `Converter(force_update=...)` → + `self._force_update_all`): every written item carries a `KP2BW_SYNC` plain-text field holding + `_content_signature(item)` — a sha256 over exactly the surface `_content_differs` compares (name, notes, + `_fields_signature` with managed stamps excluded, `_login_signature`), stamped in `_add_bw_entry_to_entries_dict`, + read by `bw_serve.item_kp2bw_sync`, and itself excluded from the signature (in `_MANAGED_FIELD_NAMES`) so it never + causes a spurious diff. `_reconcile_existing_item` computes `_is_user_modified(existing)` = + `stamp != _content_signature(existing)`; when content differs **and** the item was user-edited **and** not + `force_update`/`_force_update_all`, it returns `"protected"` (no PUT, attachments skipped) instead of clobbering. + kp2bw's own writes restamp, so they never self-trip; an unstamped (legacy/first-run) item returns `False` and updates + normally to establish the stamp. The signature mechanism is deliberate over comparing Bitwarden's `revisionDate` to a + client timestamp — the server bumps `revisionDate` *after* kp2bw generates a stamp, so a timestamp compare would + false-trip every kp2bw-written item. The `protected` count is reported in the summary. Tests: `protect_edits_test.py`. +- `--strip-ids` (`KP2BW_STRIP_IDS`, default off) is the finalize-mode inverse of the stamps above: it short-circuits migration entirely (`main()` returns before any KeePass read or password prompt) and only touches Bitwarden, removing - the `KP2BW_ID` field from every in-scope item via `_run_strip_ids` (`cli.py`) → `strip_field_from_items` - (`bw_serve.py`, one full `update_item` `PUT` per stamped item). Scope mirrors a migration (`-o`/`-c`), so only items - kp2bw could have stamped are touched. It is **irreversible** and degrades future re-runs (they fall back to - `(folder, name)` matching), so an interactive run confirms first (skippable with `-y`/`KP2BW_YES`); a declined prompt - exits `0` (clean abort), Ctrl+C exits `130`. Re-runnable: a second pass finds nothing. + the `KP2BW_ID` **and** `KP2BW_SYNC` fields from every in-scope item via `_run_strip_ids` (`cli.py`) → + `strip_field_from_items(*field_names)` (`bw_serve.py`, one full `update_item` `PUT` per stamped item). Scope mirrors a + migration (`-o`/`-c`), so only items kp2bw could have stamped are touched. It is **irreversible** and degrades future + re-runs (they fall back to `(folder, name)` matching), so an interactive run confirms first (skippable with + `-y`/`KP2BW_YES`); a declined prompt exits `0` (clean abort), Ctrl+C exits `130`. Re-runnable: a second pass finds + nothing. - `--metadata` (default on) folds the KeePass metadata Bitwarden has no native slot for — **tags and expiry** — into a single readable **YAML** `KP2BW_META` text field (`_build_metadata_field`, via PyYAML `safe_dump(allow_unicode= diff --git a/src/kp2bw/bw_serve.py b/src/kp2bw/bw_serve.py index 1b7b41f..34964d4 100644 --- a/src/kp2bw/bw_serve.py +++ b/src/kp2bw/bw_serve.py @@ -90,6 +90,14 @@ # collapse onto one item, and re-runs stay idempotent across title/folder edits. KP2BW_ID_FIELD_NAME: str = "KP2BW_ID" +# Hidden custom-field name carrying a content signature of what kp2bw last wrote +# to the item -- the basis for protecting Bitwarden-side manual edits on re-run. +# A re-run that finds the item's current managed content no longer matching this +# stamp knows a *user* edited it (kp2bw's own writes restamp it), and preserves +# the edit instead of clobbering it. Excluded from the content diff, like +# KP2BW_ID, so it never makes a re-run look "changed". +KP2BW_SYNC_FIELD_NAME: str = "KP2BW_SYNC" + def item_kp2bw_id(item: BwItemResponse) -> str | None: """Return *item*'s KeePass-UUID stamp, or ``None`` if it is unstamped. @@ -104,6 +112,19 @@ def item_kp2bw_id(item: BwItemResponse) -> str | None: return None +def item_kp2bw_sync(item: BwItemResponse) -> str | None: + """Return *item*'s ``KP2BW_SYNC`` content-signature stamp, or ``None``. + + Absent on legacy items and on anything kp2bw has not written since the + feature shipped; such items are treated as un-protected (the next write + establishes the stamp). See :meth:`Converter._is_user_modified`. + """ + for field in item.get("fields") or []: + if field.get("name") == KP2BW_SYNC_FIELD_NAME: + return field.get("value") or None + return None + + class StripResult(NamedTuple): """Outcome of a :meth:`BitwardenServeClient.strip_field_from_items` pass. @@ -985,16 +1006,17 @@ def update_item(self, item_id: str, item: BwItemResponse) -> None: self._request("PUT", f"/object/item/{item_id}", json_body=body) logger.log(VERBOSE, f"Updated item {item.get('name', '?')!r} ({item_id})") - def strip_field_from_items(self, field_name: str) -> StripResult: - """Remove a custom field from every in-scope item that carries it. + def strip_field_from_items(self, *field_names: str) -> StripResult: + """Remove the named custom field(s) from every in-scope item carrying any. - The finalize step for users adopting Bitwarden: drops kp2bw's - ``KP2BW_ID`` dedup stamp once a migration is trusted, leaving clean - items behind. Scope mirrors a migration -- the configured - organisation/collection when set, otherwise the personal vault -- so - only items kp2bw could have stamped are touched. Items lacking the - field are skipped; each match is rewritten via a full :meth:`update_item` - ``PUT``. Returns the scanned/stripped counts for the caller to report. + The finalize step for users adopting Bitwarden: drops kp2bw's managed + stamps (``KP2BW_ID`` dedup key and the ``KP2BW_SYNC`` edit-protection + signature) once a migration is trusted, leaving clean items behind. + Scope mirrors a migration -- the configured organisation/collection when + set, otherwise the personal vault -- so only items kp2bw could have + stamped are touched. An item carrying none of *field_names* is skipped; + each match is rewritten once via a full :meth:`update_item` ``PUT`` that + drops every named field present. Returns the scanned/stripped counts. The strip itself is re-runnable (a second pass finds nothing), but it is **irreversible** and degrades future migrations: the stamp is the stable @@ -1004,6 +1026,7 @@ def strip_field_from_items(self, field_name: str) -> StripResult: It is therefore a deliberate final step, gated by a confirmation in the CLI (skippable with ``-y`` for callers who know what they want). """ + targets = frozenset(field_names) items = self.list_items( organization_id=self._org_id, collection_id=self._collection_id, @@ -1011,15 +1034,16 @@ def strip_field_from_items(self, field_name: str) -> StripResult: stripped = 0 for item in items: fields = item.get("fields") or [] - if not any(field.get("name") == field_name for field in fields): + if not any(field.get("name") in targets for field in fields): continue item["fields"] = [ - field for field in fields if field.get("name") != field_name + field for field in fields if field.get("name") not in targets ] self.update_item(item["id"], item) stripped += 1 logger.info( - f"Stripped {field_name} from {stripped} of {len(items)} scanned items" + f"Stripped {', '.join(field_names)} from {stripped} of " + f"{len(items)} scanned items" ) return StripResult(scanned=len(items), stripped=stripped) diff --git a/src/kp2bw/cli.py b/src/kp2bw/cli.py index ea00d07..2145487 100644 --- a/src/kp2bw/cli.py +++ b/src/kp2bw/cli.py @@ -13,7 +13,12 @@ from . import VERBOSE, __title__, __version__ from ._console import console -from .bw_serve import KP2BW_ID_FIELD_NAME, BitwardenServeClient, ensure_bw_available +from .bw_serve import ( + KP2BW_ID_FIELD_NAME, + KP2BW_SYNC_FIELD_NAME, + BitwardenServeClient, + ensure_bw_available, +) from .convert import Converter, collect_keepass_uris from .exceptions import BitwardenClientError, ConversionError from .uri_mapping import ( @@ -230,6 +235,18 @@ def _argparser() -> MyArgParser: action=BooleanOptionalAction, default=None, ) + parser.add_argument( + "--force-update", + dest="force_update", + help=( + "Overwrite existing Bitwarden entries with KeePass content even when " + "they were edited in Bitwarden since the last run. By default such " + "manually-edited items are protected (skipped) so your edits survive; " + "this makes KeePass win regardless (env: KP2BW_FORCE_UPDATE)" + ), + action="store_true", + default=None, + ) parser.add_argument( "--include-oversize-secrets", dest="include_oversize_secrets", @@ -378,31 +395,37 @@ def _run_strip_ids( collection_id: str | None, skip_confirm: bool, ) -> None: - """Remove kp2bw's ``KP2BW_ID`` dedup stamp from every migrated item, then stop. + """Remove kp2bw's ``KP2BW_ID``/``KP2BW_SYNC`` stamps from every item, then stop. The finalize step for users ready to fully adopt Bitwarden: no KeePass database is read, and scope follows ``-o``/``-c`` exactly as a migration - would. The operation is **irreversible** -- the stamp cannot be recovered -- - and makes future migration re-runs unreliable: without it, entries that share - a folder + name (the very collision the stamp disambiguates) can be - duplicated or mismatched. A loud confirmation states this before any change; - ``-y``/``--yes`` skips it for callers who know what they want. Errors surface - as an actionable message rather than a traceback. + would. Drops both the ``KP2BW_ID`` dedup key and the ``KP2BW_SYNC`` edit- + protection signature. The operation is **irreversible** -- the stamps cannot + be recovered -- and makes future migration re-runs unreliable: without + ``KP2BW_ID``, entries that share a folder + name (the very collision it + disambiguates) can be duplicated or mismatched, and without ``KP2BW_SYNC`` + manual-edit protection no longer applies. A loud confirmation states this + before any change; ``-y``/``--yes`` skips it for callers who know what they + want. Errors surface as an actionable message rather than a traceback. """ scope = _describe_scope(org_id, collection_id) try: if not skip_confirm: console.print( f"[bold yellow]Warning:[/bold yellow] this removes the " - f"[bold]{KP2BW_ID_FIELD_NAME}[/bold] field from every " + f"[bold]{KP2BW_ID_FIELD_NAME}[/bold] and " + f"[bold]{KP2BW_SYNC_FIELD_NAME}[/bold] fields from every " f"kp2bw-migrated item in [bold]{escape(scope)}[/bold]. Other " "vault data is untouched, but this [bold]cannot be undone[/bold] " "and makes future migration re-runs unreliable: without the " - "stamp, entries sharing a folder + name can be duplicated or " - "mismatched. Only do this once you are finished migrating and " - "ready to fully adopt Bitwarden." + "stamps, entries sharing a folder + name can be duplicated or " + "mismatched and manual-edit protection no longer applies. Only do " + "this once you are finished migrating and ready to fully adopt " + "Bitwarden." ) - if not _confirm(f"Remove {KP2BW_ID_FIELD_NAME} stamps? [y/n]: "): + if not _confirm( + f"Remove {KP2BW_ID_FIELD_NAME}/{KP2BW_SYNC_FIELD_NAME} stamps? [y/n]: " + ): console.print("[yellow]Aborted; nothing changed.[/yellow]") sys.exit(0) @@ -412,7 +435,9 @@ def _run_strip_ids( with BitwardenServeClient( bw_pw, org_id=org_id, collection_id=collection_id ) as bw: - result = bw.strip_field_from_items(KP2BW_ID_FIELD_NAME) + result = bw.strip_field_from_items( + KP2BW_ID_FIELD_NAME, KP2BW_SYNC_FIELD_NAME + ) except KeyboardInterrupt: console.print("\n[yellow]Interrupted.[/yellow]") sys.exit(130) @@ -421,13 +446,13 @@ def _run_strip_ids( if result.stripped: console.print( - f"[green]Removed {KP2BW_ID_FIELD_NAME} from {result.stripped} of " - f"{result.scanned} item(s).[/green]" + f"[green]Removed {KP2BW_ID_FIELD_NAME}/{KP2BW_SYNC_FIELD_NAME} from " + f"{result.stripped} of {result.scanned} item(s).[/green]" ) else: console.print( - f"No items carried a {KP2BW_ID_FIELD_NAME} stamp in {escape(scope)} " - f"({result.scanned} scanned); nothing to do." + f"No items carried a {KP2BW_ID_FIELD_NAME}/{KP2BW_SYNC_FIELD_NAME} " + f"stamp in {escape(scope)} ({result.scanned} scanned); nothing to do." ) @@ -702,6 +727,9 @@ def main() -> None: update_existing = _resolve_bool_option( args.update_existing, "KP2BW_UPDATE", default=True ) + force_update = _resolve_bool_option( + args.force_update, "KP2BW_FORCE_UPDATE", default=False + ) include_oversize_secrets = _resolve_bool_option( args.include_oversize_secrets, "KP2BW_INCLUDE_OVERSIZE_SECRETS", @@ -883,6 +911,7 @@ def main() -> None: include_recyclebin=include_recyclebin, migrate_metadata=migrate_metadata, update_existing=update_existing, + force_update=force_update, include_oversize_secrets=include_oversize_secrets, uri_match=uri_match, interpret_uri_syntax=interpret_uri_syntax, diff --git a/src/kp2bw/convert.py b/src/kp2bw/convert.py index 8026c19..d4fc0bf 100644 --- a/src/kp2bw/convert.py +++ b/src/kp2bw/convert.py @@ -1,6 +1,7 @@ import base64 import binascii import copy +import hashlib import logging import time from itertools import islice @@ -20,7 +21,12 @@ from . import VERBOSE from ._console import console -from .bw_serve import KP2BW_ID_FIELD_NAME, BitwardenServeClient +from .bw_serve import ( + KP2BW_ID_FIELD_NAME, + KP2BW_SYNC_FIELD_NAME, + BitwardenServeClient, + item_kp2bw_sync, +) from .bw_types import ( BwFido2Credential, BwField, @@ -69,6 +75,7 @@ def _print_summary( n_created: int, n_updated: int, n_skipped: int, + n_protected: int, n_collection_update: int, n_attachments: int, n_update_failed: int, @@ -85,6 +92,7 @@ def _print_summary( n_created, n_updated, n_skipped, + n_protected, n_collection_update, n_attachments, n_update_failed, @@ -99,6 +107,11 @@ def _print_summary( console.print(f" [blue]{n_updated:{w}d}[/blue] updated (changed in KeePass)") if n_skipped: console.print(f" [dim]{n_skipped:{w}d}[/dim] skipped (unchanged)") + if n_protected: + console.print( + f" [magenta]{n_protected:{w}d}[/magenta] protected (edited in " + f"Bitwarden; use --force-update to overwrite)" + ) if n_collection_update: console.print( f" [yellow]{n_collection_update:{w}d}[/yellow] added to collection" @@ -221,6 +234,7 @@ def __init__( include_recyclebin: bool = False, migrate_metadata: bool = True, update_existing: bool = True, + force_update: bool = False, include_oversize_secrets: bool = False, uri_match: UriMatchValue = None, interpret_uri_syntax: bool = True, @@ -239,6 +253,7 @@ def __init__( self._include_recyclebin = include_recyclebin self._migrate_metadata = migrate_metadata self._update_existing = update_existing + self._force_update_all = force_update self._include_oversize_secrets = include_oversize_secrets self._uri_match = uri_match self._interpret_uri_syntax = interpret_uri_syntax @@ -515,6 +530,20 @@ def _add_bw_entry_to_entries_dict( android_packages=[v for _, v in sorted(app_attrs)], ) + # Stamp the content signature kp2bw is writing, so a later re-run can tell + # a user's manual Bitwarden edit (signature no longer matches) from kp2bw's + # own prior write (this restamps it). Plain text, not hidden: like + # KP2BW_ID it is metadata, not a secret, and a hidden field is only + # UI-masked, not protected. Excluded from the content signature itself, so + # it never makes a re-run look "changed". + bw_item_object["fields"].append( + BwField( + name=KP2BW_SYNC_FIELD_NAME, + value=self._content_signature(bw_item_object), + type=0, + ) + ) + # get attachments to store later on. A value over the inline size limit # is offloaded to a .txt attachment, with three exceptions: # * consumed OTP keys are already folded into login.totp, so dropping @@ -921,45 +950,72 @@ def _materialise_attachment(att: AttachmentItem) -> tuple[str, bytes]: return name, att.data return name, att[1].encode("UTF-8") - @staticmethod + # Custom fields kp2bw manages itself, never user content: excluded from the + # content signature so they can never make a re-run look "changed". + _MANAGED_FIELD_NAMES: frozenset[str] = frozenset({ + KP2BW_ID_FIELD_NAME, + KP2BW_SYNC_FIELD_NAME, + }) + + @classmethod def _fields_signature( + cls, fields: list[BwField] | None, ) -> list[tuple[str, str, int]]: """Order-independent (name, value, type) signature of custom fields. - The ``KP2BW_ID`` identity stamp is excluded: it is kp2bw-managed metadata, - not user content, so it must never make a re-run look "changed" (and on a - legacy item it is absent, while the desired item always carries it). + The kp2bw-managed stamps (``KP2BW_ID`` identity, ``KP2BW_SYNC`` sync + signature) are excluded: they are metadata, not user content, so they + must never make a re-run look "changed" (and on a legacy item they are + absent, while the desired item always carries ``KP2BW_ID``). """ return sorted( ( (f.get("name") or "", f.get("value") or "", f.get("type") or 0) for f in (fields or []) - if (f.get("name") or "") != KP2BW_ID_FIELD_NAME + if (f.get("name") or "") not in cls._MANAGED_FIELD_NAMES ), key=lambda t: (t[0], t[2], t[1]), ) @staticmethod - def _login_differs(existing: BwItemLogin | None, desired: BwItemLogin) -> bool: + def _login_signature( + login: BwItemLogin | None, + ) -> tuple[str, str, str, list[str]]: + """Signature of the login fields kp2bw owns (creds, totp, URIs).""" + if login is None: + return ("", "", "", []) + return ( + login.get("username") or "", + login.get("password") or "", + login.get("totp") or "", + [u.get("uri", "") for u in (login.get("uris") or [])], + ) + + @classmethod + def _login_differs(cls, existing: BwItemLogin | None, desired: BwItemLogin) -> bool: """Compare the login fields kp2bw owns (creds, totp, URIs).""" - if existing is None: - existing = BwItemLogin( - uris=[], - username="", - password="", - totp=None, - passwordRevisionDate=None, - ) - if (existing.get("username") or "") != (desired.get("username") or ""): - return True - if (existing.get("password") or "") != (desired.get("password") or ""): - return True - if (existing.get("totp") or "") != (desired.get("totp") or ""): - return True - ex_uris = [u.get("uri", "") for u in (existing.get("uris") or [])] - de_uris = [u.get("uri", "") for u in (desired.get("uris") or [])] - return ex_uris != de_uris + return cls._login_signature(existing) != cls._login_signature(desired) + + @classmethod + def _content_signature(cls, item: BwItemResponse | BwItemCreate) -> str: + """Hex digest over exactly the content kp2bw manages on an item. + + Covers name, notes, the custom-field signature (managed stamps excluded) + and the login signature -- the same surface :meth:`_content_differs` + compares, so the value stamped into ``KP2BW_SYNC`` on a write and the + value recomputed on the next run agree whenever a user has not touched + the item. A user edit to any of these flips the digest; a kp2bw write + restamps it. The digest leaks nothing: every input is already present in + cleartext on the same item to anyone who can read it. + """ + blob = repr(( + item.get("name") or "", + item.get("notes") or "", + cls._fields_signature(item.get("fields")), + cls._login_signature(item.get("login")), + )) + return hashlib.sha256(blob.encode("utf-8")).hexdigest() @classmethod def _content_differs(cls, existing: BwItemResponse, desired: BwItemCreate) -> bool: @@ -969,15 +1025,23 @@ def _content_differs(cls, existing: BwItemResponse, desired: BwItemCreate) -> bo the login credentials/URIs) so an unchanged re-run stays idempotent and never issues a redundant PUT. """ - if (existing.get("name") or "") != (desired.get("name") or ""): - return True - if (existing.get("notes") or "") != (desired.get("notes") or ""): - return True - if cls._fields_signature(existing.get("fields")) != cls._fields_signature( - desired.get("fields") - ): - return True - return cls._login_differs(existing.get("login"), desired["login"]) + return cls._content_signature(existing) != cls._content_signature(desired) + + @classmethod + def _is_user_modified(cls, existing: BwItemResponse) -> bool: + """True if a user edited *existing*'s managed content since kp2bw's write. + + Compares the item's current content signature against the ``KP2BW_SYNC`` + stamp kp2bw wrote last time. A mismatch means the managed content + changed outside kp2bw (kp2bw restamps on every write, so its own updates + never trip this). An unstamped item -- legacy, or not yet written since + the feature shipped -- returns ``False`` so the next run establishes the + stamp rather than freezing it behind ``--force-update``. + """ + stamp = item_kp2bw_sync(existing) + if stamp is None: + return False + return stamp != cls._content_signature(existing) @staticmethod def _build_update_payload( @@ -1081,7 +1145,7 @@ def _reconcile_existing_item( kp_uuid: str, force_update: bool = False, ) -> tuple[ - Literal["updated", "collection", "skipped", "failed"], + Literal["updated", "collection", "skipped", "protected", "failed"], list[AttachmentItem], dict[str, str], ]: @@ -1089,7 +1153,8 @@ def _reconcile_existing_item( Returns ``(outcome, upload_attachments, stale_by_name)`` where *outcome* is one of ``"updated"`` (content PUT), ``"collection"`` (membership-only - PUT), ``"skipped"`` (no change) or ``"failed"`` (the PUT was rejected); + PUT), ``"skipped"`` (no change), ``"protected"`` (a manual Bitwarden edit + preserved instead of clobbered) or ``"failed"`` (the PUT was rejected); *upload_attachments* are the files to (re-)upload -- those the item does not have yet plus those whose content changed; and *stale_by_name* maps a changed file's name to the id of the stale copy to delete once its @@ -1100,10 +1165,18 @@ def _reconcile_existing_item( content is unchanged -- used when adopting a legacy item so its missing ``KP2BW_ID`` stamp is backfilled. It still respects ``--no-update`` (the PUT is gated by ``self._update_existing``). + + When the item was edited in Bitwarden since kp2bw last wrote it (its + ``KP2BW_SYNC`` signature no longer matches), the overwrite is skipped and + ``"protected"`` returned so the manual edit survives -- unless + ``self._force_update_all`` (``--force-update``) makes KeePass win. Legacy + adoption (*force_update*) and unstamped items are never protected. """ name = bw_item["name"] item_id = existing["id"] - outcome: Literal["updated", "collection", "skipped", "failed"] = "skipped" + outcome: Literal["updated", "collection", "skipped", "protected", "failed"] = ( + "skipped" + ) # kp2bw only ever creates login items, so a non-login vault item sharing # this (folder, name) is a name collision we must not mutate -- neither @@ -1121,9 +1194,26 @@ def _reconcile_existing_item( try: # Content sync: PUT only when the KeePass-derived content changed # (keeps re-runs idempotent). - if self._update_existing and ( - force_update or self._content_differs(existing, bw_item) - ): + content_differs = self._content_differs(existing, bw_item) + if self._update_existing and (force_update or content_differs): + # Protect a manual Bitwarden edit: when the content genuinely + # diverged (not a forced legacy-adoption PUT) and the item was + # touched in Bitwarden since kp2bw last wrote it, preserve the + # edit instead of clobbering it -- unless --force-update makes + # KeePass win. Skip attachments too: a protected item is left + # wholly untouched rather than half-synced. + if ( + content_differs + and not force_update + and not self._force_update_all + and self._is_user_modified(existing) + ): + logger.warning( + f"-- Entry {name!r}: modified in Bitwarden since the last " + f"sync; preserving your edit (use --force-update to " + f"overwrite with KeePass)" + ) + return "protected", [], {} payload = self._build_update_payload(existing, bw_item) bw.update_item(item_id, payload) bw.update_dedup_entry(kp_uuid, payload) @@ -1208,6 +1298,7 @@ def _create_bitwarden_items_for_entries(self) -> int: n_skipped = 0 n_updated = 0 + n_protected = 0 n_collection_update = 0 n_created = 0 n_create_failed = 0 @@ -1290,6 +1381,8 @@ def _create_bitwarden_items_for_entries(self) -> int: n_updated += 1 elif outcome == "collection": n_collection_update += 1 + elif outcome == "protected": + n_protected += 1 elif outcome == "failed": n_update_failed += 1 else: # "skipped" — content unchanged (attachments, if any, @@ -1411,6 +1504,7 @@ def _on_create_failed(_key: str, _exc: BitwardenClientError) -> None: n_created, n_updated, n_skipped, + n_protected, n_collection_update, n_attachments, n_update_failed, diff --git a/tests/AGENTS.md b/tests/AGENTS.md index 5721b30..edc4d64 100644 --- a/tests/AGENTS.md +++ b/tests/AGENTS.md @@ -6,20 +6,21 @@ Tests are script-executable smoke/e2e checks, with dockerized Vaultwarden integr ## WHERE TO LOOK -| Task | Location | Notes | -| ------------------------- | -------------------------------- | -------------------------------------------------------- | -| bw command resolution | `tests/bw_serve_command_test.py` | Cross-platform argv wrapping + process teardown logic | -| Windows bw.cmd live smoke | `tests/windows_bw_cmd_smoke.py` | Gated by `KP2BW_RUN_WIN_CMD_SMOKE=1`; Windows CI only | -| Package smoke validation | `tests/smoke_test.py` | Verifies built artifact behavior, metadata, entry points | -| Stubs package validation | `tests/stubs_smoke_test.py` | Verifies `partial` marker + type-check consumption | -| Full migration e2e | `tests/e2e_vaultwarden_test.py` | Rich seed + idempotency + golden snapshots | -| Snapshot normalization | `tests/_snapshot.py` | Scrubs volatile fields, hashes attachments, golden diff | -| Golden snapshots | `tests/__snapshots__/*.json` | Committed expected vault state (pinned bw owns them) | -| Integration infra | `tests/docker-compose.yml` | Vaultwarden + test container orchestration | -| Vaultwarden image seeding | `tests/Dockerfile.vaultwarden` | Pinned image + DB/key fixtures | -| Test runner image | `tests/Dockerfile.test` | uv + lockfile-pinned bw CLI + test deps | -| Reusable bw setup | `.github/actions/setup-bw` | Installs a chosen `@bitwarden/cli` (matrix-able) | -| Fixture retention rules | `tests/fixtures/.gitignore` | Allowlist keeps required DB/certs tracked | +| Task | Location | Notes | +| ------------------------- | -------------------------------- | --------------------------------------------------------- | +| bw command resolution | `tests/bw_serve_command_test.py` | Cross-platform argv wrapping + process teardown logic | +| Manual-edit protection | `tests/protect_edits_test.py` | `KP2BW_SYNC` stamp, `_is_user_modified`, protect vs force | +| Windows bw.cmd live smoke | `tests/windows_bw_cmd_smoke.py` | Gated by `KP2BW_RUN_WIN_CMD_SMOKE=1`; Windows CI only | +| Package smoke validation | `tests/smoke_test.py` | Verifies built artifact behavior, metadata, entry points | +| Stubs package validation | `tests/stubs_smoke_test.py` | Verifies `partial` marker + type-check consumption | +| Full migration e2e | `tests/e2e_vaultwarden_test.py` | Rich seed + idempotency + golden snapshots | +| Snapshot normalization | `tests/_snapshot.py` | Scrubs volatile fields, hashes attachments, golden diff | +| Golden snapshots | `tests/__snapshots__/*.json` | Committed expected vault state (pinned bw owns them) | +| Integration infra | `tests/docker-compose.yml` | Vaultwarden + test container orchestration | +| Vaultwarden image seeding | `tests/Dockerfile.vaultwarden` | Pinned image + DB/key fixtures | +| Test runner image | `tests/Dockerfile.test` | uv + lockfile-pinned bw CLI + test deps | +| Reusable bw setup | `.github/actions/setup-bw` | Installs a chosen `@bitwarden/cli` (matrix-able) | +| Fixture retention rules | `tests/fixtures/.gitignore` | Allowlist keeps required DB/certs tracked | ## CONVENTIONS diff --git a/tests/_snapshot.py b/tests/_snapshot.py index 1d9a0f9..5accb9e 100644 --- a/tests/_snapshot.py +++ b/tests/_snapshot.py @@ -30,7 +30,7 @@ from pathlib import Path from typing import TypedDict -from kp2bw.bw_serve import KP2BW_ID_FIELD_NAME +from kp2bw.bw_serve import KP2BW_ID_FIELD_NAME, KP2BW_SYNC_FIELD_NAME # A concrete recursive type for parsed JSON. Laundering ``json.loads``'s ``Any`` # into this once (``as_object`` / ``parse_object``) means every later @@ -247,10 +247,10 @@ def _norm_item( ( _norm_field(f) for f in _objects(raw, "fields") - # kp2bw's stable-identity stamp is volatile per run (a KeePass UUID) - # and is implementation metadata, not migrated content -- drop it so - # the golden stays deterministic across runs. - if f.get("name") != KP2BW_ID_FIELD_NAME + # kp2bw's managed stamps are volatile per run (KP2BW_ID is a KeePass + # UUID; KP2BW_SYNC is a content hash) and are implementation metadata, + # not migrated content -- drop them so the golden stays deterministic. + if f.get("name") not in (KP2BW_ID_FIELD_NAME, KP2BW_SYNC_FIELD_NAME) ), key=lambda f: (f["name"] or "", f["type"], f["value"] or ""), ) diff --git a/tests/protect_edits_test.py b/tests/protect_edits_test.py new file mode 100644 index 0000000..5a75a0e --- /dev/null +++ b/tests/protect_edits_test.py @@ -0,0 +1,254 @@ +"""Manual-edit protection on re-runs (issue #30). + +`Converter` stamps a `KP2BW_SYNC` content signature on every item it writes. A +re-run that finds the stamp no longer matching the item's current managed +content knows a *user* edited it in Bitwarden and preserves the edit (outcome +`"protected"`) instead of clobbering it -- unless `--force-update` makes KeePass +win. kp2bw's own writes restamp, so they never self-trip the protection. The +stamp is excluded from the content signature, so it never causes a spurious +diff. Driven with a client double, so no live `bw serve` process is spawned. +""" + +from typing import cast + +from kp2bw.bw_serve import ( + KP2BW_ID_FIELD_NAME, + KP2BW_SYNC_FIELD_NAME, + BitwardenServeClient, +) +from kp2bw.bw_types import BwField, BwItemCreate, BwItemResponse +from kp2bw.convert import Converter + +_BW_LOGIN = 1 + + +def _make_converter(*, force_update: bool = False) -> Converter: + """Build a converter against dummy credentials; no I/O happens in __init__.""" + return Converter( + keepass_file_path="dummy.kdbx", + keepass_password="pw", + keepass_keyfile_path=None, + bitwarden_password="pw", + bitwarden_organization_id=None, + bitwarden_coll_id=None, + path2name=False, + path2nameskip=1, + import_tags=None, + force_update=force_update, + ) + + +def _field(name: str, value: str) -> BwField: + return cast(BwField, {"name": name, "value": value, "type": 0}) + + +def _desired(*, name: str, password: str, note: str = "") -> BwItemCreate: + """A freshly-built item as the migration would emit it (KeePass-authoritative).""" + return cast( + BwItemCreate, + { + "organizationId": None, + "collectionIds": [], + "folderId": None, + "type": _BW_LOGIN, + "name": name, + "notes": note, + "favorite": False, + "fields": [_field(KP2BW_ID_FIELD_NAME, "UUID")], + "login": { + "uris": [], + "username": "u", + "password": password, + "totp": None, + "passwordRevisionDate": None, + }, + "secureNote": None, + "card": None, + "identity": None, + }, + ) + + +def _existing_as_kp2bw_wrote_it(*, name: str, password: str) -> BwItemResponse: + """An existing vault item carrying the KP2BW_SYNC stamp kp2bw last wrote.""" + item = cast( + BwItemResponse, + { + "id": "item-id", + "object": "item", + "revisionDate": "2026-01-01T00:00:00.000Z", + "organizationId": None, + "collectionIds": [], + "folderId": None, + "type": _BW_LOGIN, + "name": name, + "notes": "", + "favorite": False, + "fields": [_field(KP2BW_ID_FIELD_NAME, "UUID")], + "login": { + "uris": [], + "username": "u", + "password": password, + "totp": None, + "passwordRevisionDate": None, + }, + "secureNote": None, + "card": None, + "identity": None, + }, + ) + # Stamp it exactly as _add_bw_entry_to_entries_dict would: the signature of + # the content kp2bw wrote, excluded from that same signature. + item["fields"].append( + _field(KP2BW_SYNC_FIELD_NAME, Converter._content_signature(item)) + ) + return item + + +class _FakeBw: + """Records every update_item PUT so a protected item proves it issued none.""" + + def __init__(self) -> None: + self.updated: list[str] = [] + + def update_item(self, item_id: str, item: BwItemResponse) -> None: + self.updated.append(item_id) + + def update_dedup_entry(self, kp_uuid: str, item: BwItemResponse) -> None: + pass + + +def _reconcile( + converter: Converter, existing: BwItemResponse, desired: BwItemCreate +) -> tuple[str, _FakeBw]: + bw = _FakeBw() + outcome, _uploads, _stale = converter._reconcile_existing_item( + cast(BitwardenServeClient, bw), + existing, + None, + desired, + [], + fixed_coll_id=None, + kp_uuid="UUID", + force_update=False, + ) + return outcome, bw + + +def assert_sync_stamp_excluded_from_signature() -> None: + """The KP2BW_SYNC stamp must not change the content signature it records.""" + desired = _desired(name="X", password="p") + sig = Converter._content_signature(desired) + stamped = cast(BwItemCreate, dict(desired)) + stamped["fields"] = [ + *desired["fields"], + _field(KP2BW_SYNC_FIELD_NAME, sig), + ] + if Converter._content_signature(stamped) != sig: + raise AssertionError("KP2BW_SYNC stamp must not affect the content signature") + + +def assert_legacy_item_not_user_modified() -> None: + """An unstamped (legacy) item is never treated as user-modified.""" + existing = _existing_as_kp2bw_wrote_it(name="X", password="p") + existing["fields"] = [ + f for f in existing["fields"] if f.get("name") != KP2BW_SYNC_FIELD_NAME + ] + if Converter._is_user_modified(existing): + raise AssertionError("an unstamped legacy item must not be protected") + + +def assert_own_write_not_user_modified() -> None: + """A freshly kp2bw-stamped item must not look user-modified (no self-trip).""" + existing = _existing_as_kp2bw_wrote_it(name="X", password="p") + if Converter._is_user_modified(existing): + raise AssertionError("kp2bw's own write must not trip the protection") + + +def assert_user_edit_detected() -> None: + """A Bitwarden-side edit flips the signature, so it is detected.""" + existing = _existing_as_kp2bw_wrote_it(name="X", password="orig") + login = existing.get("login") + assert login is not None + login["password"] = "user-edited" + if not Converter._is_user_modified(existing): + raise AssertionError("a manual Bitwarden edit must be detected") + + +def assert_protected_when_user_edited() -> None: + """User edit + KeePass change + no force -> protected, no PUT issued.""" + existing = _existing_as_kp2bw_wrote_it(name="X", password="orig") + login = existing.get("login") + assert login is not None + login["password"] = "user-edited" + desired = _desired(name="X", password="kp-new") + + outcome, bw = _reconcile(_make_converter(), existing, desired) + + if outcome != "protected": + raise AssertionError(f"expected 'protected', got {outcome!r}") + if bw.updated: + raise AssertionError(f"a protected item must not be PUT: {bw.updated}") + + +def assert_force_update_overwrites_user_edit() -> None: + """--force-update overrides protection: KeePass wins, item is PUT.""" + existing = _existing_as_kp2bw_wrote_it(name="X", password="orig") + login = existing.get("login") + assert login is not None + login["password"] = "user-edited" + desired = _desired(name="X", password="kp-new") + + outcome, bw = _reconcile(_make_converter(force_update=True), existing, desired) + + if outcome != "updated": + raise AssertionError( + f"expected 'updated' under --force-update, got {outcome!r}" + ) + if bw.updated != ["item-id"]: + raise AssertionError(f"forced update must PUT the item: {bw.updated}") + + +def assert_keepass_change_updates_unedited_item() -> None: + """KeePass changed but the user did not touch Bitwarden -> normal update.""" + existing = _existing_as_kp2bw_wrote_it(name="X", password="orig") + desired = _desired(name="X", password="kp-new") + + outcome, bw = _reconcile(_make_converter(), existing, desired) + + if outcome != "updated": + raise AssertionError(f"an unedited item should update, got {outcome!r}") + if bw.updated != ["item-id"]: + raise AssertionError(f"the update must PUT the item: {bw.updated}") + + +def assert_matching_content_skips_regardless_of_edit() -> None: + """User edit that already matches KeePass content needs no PUT -> skipped.""" + existing = _existing_as_kp2bw_wrote_it(name="X", password="orig") + login = existing.get("login") + assert login is not None + login["password"] = "shared" + desired = _desired(name="X", password="shared") + + outcome, bw = _reconcile(_make_converter(), existing, desired) + + if outcome != "skipped": + raise AssertionError(f"matching content should skip, got {outcome!r}") + if bw.updated: + raise AssertionError(f"nothing to change, so no PUT: {bw.updated}") + + +def main() -> None: + assert_sync_stamp_excluded_from_signature() + assert_legacy_item_not_user_modified() + assert_own_write_not_user_modified() + assert_user_edit_detected() + assert_protected_when_user_edited() + assert_force_update_overwrites_user_edit() + assert_keepass_change_updates_unedited_item() + assert_matching_content_skips_regardless_of_edit() + print("protect edits test passed") + + +if __name__ == "__main__": + main() diff --git a/tests/strip_ids_test.py b/tests/strip_ids_test.py index a31bf7d..ecae9ee 100644 --- a/tests/strip_ids_test.py +++ b/tests/strip_ids_test.py @@ -143,7 +143,7 @@ def __enter__(self) -> Self: def __exit__(self, *_exc: object) -> None: return None - def strip_field_from_items(self, _field_name: str) -> StripResult: + def strip_field_from_items(self, *_field_names: str) -> StripResult: return StripResult(scanned=3, stripped=0) def _must_not_prompt(*_args: object, **_kwargs: object) -> str: diff --git a/tests/test_script_adapters.py b/tests/test_script_adapters.py index 5d4534c..ec25f80 100644 --- a/tests/test_script_adapters.py +++ b/tests/test_script_adapters.py @@ -55,6 +55,10 @@ def test_migrate_uris_script() -> None: _run_script_main("migrate_uris_test.py") +def test_protect_edits_script() -> None: + _run_script_main("protect_edits_test.py") + + def test_otp_script() -> None: _run_script_main("otp_test.py")