diff --git a/1password/migration/keepass/README.md b/1password/migration/keepass/README.md index 9dd7c19..fbdc6f4 100644 --- a/1password/migration/keepass/README.md +++ b/1password/migration/keepass/README.md @@ -1,24 +1,24 @@ # KeePass XML → 1Password Importer -Migrates a KeePass XML 2.x export into 1Password using the [onepassword-sdk](https://github.com/1Password/onepassword-sdk-python) bulk-create API. +Migrates a KeePass XML 2.x export into a single 1Password vault using the [onepassword-sdk](https://github.com/1Password/onepassword-sdk-python) bulk-create API. The vault is created by the service account if it does not already exist. ## Features -- **Password history** — each entry's historical passwords are appended to the item's Notes field (newest first, duplicates suppressed) -- **Attachments** — binary attachments from the KeePass pool are imported as 1Password file attachments +- **Single vault** — all entries land in one vault; the service account creates it if needed +- **Group → tag mapping** — KeePass group paths become tags, with one tag per level of hierarchy +- **Password history** — KeePass history is replayed as real 1Password item history (viewable via "View password history" in the UI) +- **Attachments** — binary attachments are imported as 1Password file attachments - **TOTP** — `otpauth://` URIs are mapped to a proper OTP field -- **Group → Vault mapping** — KeePass groups become 1Password vaults; sub-groups either get their own vault or collapse into the parent as tags (see `--collapse-folders`) - **Recycle Bin** — skipped automatically - **Resumable** — a state file is written on interruption (e.g. rate limit); re-running the same command picks up where it left off ## Requirements -| Requirement | Notes | -| -------------------------- | ------------------------------------------------------------ | -| Python 3.9+ | | -| `OP_SERVICE_ACCOUNT_TOKEN` | Set as an environment variable | -| `onepassword-sdk` | Auto-installed into `.venv-1pw` if missing | -| `op` CLI | Only needed for `--user-for-private` vault permission grants | +| Requirement | Notes | +| -------------------------- | ------------------------------------------ | +| Python 3.9+ | | +| `OP_SERVICE_ACCOUNT_TOKEN` | Set as an environment variable | +| `onepassword-sdk` | Auto-installed into `.venv-1pw` if missing | ## Usage @@ -28,71 +28,44 @@ export OP_SERVICE_ACCOUNT_TOKEN=your-token-here # Preview — no changes made python import-from-keepass-xml.py \ --input keepass-export.xml \ - --employee-vault "KeePass Import" \ + --vault "KeePass Import" \ --dry-run # Live import python import-from-keepass-xml.py \ --input keepass-export.xml \ - --employee-vault "KeePass Import" - -# Collapse sub-groups into parent vaults (sub-groups become tags) -python import-from-keepass-xml.py \ - --input keepass-export.xml \ - --employee-vault "KeePass Import" \ - --collapse-folders - -# Grant a user edit access to all imported vaults -python import-from-keepass-xml.py \ - --input keepass-export.xml \ - --employee-vault "KeePass Import" \ - --user-for-private user@example.com + --vault "KeePass Import" ``` ## Options -| Flag | Default | Description | -| -------------------- | ------------ | ------------------------------------------------------------- | -| `--input` | _(required)_ | Path to KeePass XML 2.x export | -| `--employee-vault` | _(required)_ | Fallback vault for entries with no group | -| `--private-prefix` | `""` | String prepended to vault names derived from groups | -| `--collapse-folders` | off | Collapse sub-groups into parent vault; sub-groups become tags | -| `--user-for-private` | — | Grant this user (email) edit access to all vaults | -| `--dry-run` | off | Print planned actions without creating anything | -| `--silent` | off | Suppress progress output | +| Flag | Description | +| ----------- | ---------------------------------------------------------------- | +| `--input` | Path to KeePass XML 2.x export _(required)_ | +| `--vault` | Destination vault name; created if it doesn't exist _(required)_ | +| `--dry-run` | Print planned actions without creating anything | +| `--silent` | Suppress progress output | -## Vault mapping +## Tag mapping -Without `--collapse-folders`, each unique group path becomes its own vault: +KeePass group paths are converted to tags, with a tag added for each level of the hierarchy so items remain discoverable by parent group: ``` -Email → vault "Email" -Email/Work → vault "Email/Work" -Development → vault "Development" -Development/Cloud → vault "Development/Cloud" +(no group) → no tags +Email → ["Email"] +Email/Work → ["Email", "Email/Work"] +Development/Cloud → ["Development", "Development/Cloud"] ``` -With `--collapse-folders`, only the top-level group is a vault; deeper paths become tags: +## Password history -``` -Email → vault "Email", tag "Email" -Email/Work → vault "Email", tag "Email\Work" -Development/Cloud → vault "Development", tag "Development\Cloud" -``` - -Entries with no group go to the `--employee-vault`. - -## Password history format +KeePass password history is replayed as real 1Password item history, visible in the UI via **View password history**: -Historical passwords are appended to the item's Notes field: - -``` ---- Password History --- -2024-01-15T09:23:00Z previous-password -2022-06-01T14:00:00Z older-password -``` +1. The item is created with the **oldest** historical password. +2. The item is updated (`put()`) once per subsequent password, oldest → newest. +3. A final `put()` restores the **current** password. -The current password is never repeated in the history block. Exact duplicate historical passwords are suppressed. +Each `put()` call produces a genuine 1Password history entry. Items without history are bulk-created normally (no extra API calls). Timestamps from KeePass are not preserved — history entries will show the time the import ran. ## Resuming an interrupted import @@ -105,6 +78,6 @@ A test XML file (`keepass-test-export.xml`) is provided. It covers logins, secur ```bash python import-from-keepass-xml.py \ --input keepass-test-export.xml \ - --employee-vault "KeePass Import" \ + --vault "KeePass Import" \ --dry-run ``` diff --git a/1password/migration/keepass/TestDatabase.xml b/1password/migration/keepass/TestDatabase.xml new file mode 100644 index 0000000..ddc1ff2 --- /dev/null +++ b/1password/migration/keepass/TestDatabase.xml @@ -0,0 +1,1299 @@ + + + + KeePass + 2026-06-16T14:29:35Z + TestDatabase + 2026-06-16T14:29:35Z + This is a test database to provide test/sample data of a keepass export + 2026-06-16T14:29:35Z + + 2026-06-16T14:28:22Z + 365 + + 2026-06-16T14:28:22Z + -1 + -1 + + False + False + True + False + False + + True + k702N2RWGk6wDtw3RHADUA== + 2026-06-16T14:28:22Z + AAAAAAAAAAAAAAAAAAAAAA== + 2026-06-16T14:28:22Z + 10 + 6291456 + jN/K/nSRM0qNnq7W20TtTg== + 7DaGTBCYPUWr1f5xmU3h5g== + + + + + + 7DaGTBCYPUWr1f5xmU3h5g== + TestDatabase + + 49 + + 2026-06-16T14:28:22Z + 2026-06-16T14:28:22Z + 2026-06-16T15:20:18Z + 2026-06-03T13:08:10Z + False + 55 + 2026-06-16T14:28:22Z + + True + + Null + Null + /56qzhmfFE6DGyzDi2ulqQ== + + /56qzhmfFE6DGyzDi2ulqQ== + 0 + + + + + + 2026-06-16T14:29:35Z + 2026-06-16T14:29:35Z + 2026-06-16T14:29:35Z + 2026-06-03T13:08:10Z + False + 0 + 2026-06-16T14:29:35Z + + + Notes + Notes + + + Password + Password + + + Title + Sample Entry + + + URL + https://keepass.info/ + + + UserName + User Name + + + True + 0 + + Target Window + {USERNAME}{TAB}{PASSWORD}{TAB}{ENTER} + + + + + + O7bys2y1IU+nUtPiD0MmOg== + 0 + + + + + + 2026-06-16T14:29:35Z + 2026-06-16T14:29:35Z + 2026-06-16T14:29:35Z + 2026-06-03T13:08:10Z + False + 0 + 2026-06-16T14:29:35Z + + + Password + 12345 + + + Title + Sample Entry #2 + + + URL + https://keepass.info/help/kb/testform.html + + + UserName + Michael321 + + + True + 0 + + *Test Form - KeePass* + + + + + + + k702N2RWGk6wDtw3RHADUA== + Recycle Bin + + 43 + + 2026-06-16T14:31:36Z + 2026-06-16T14:31:36Z + 2026-06-16T15:15:55Z + 2026-06-03T13:08:10Z + False + 7 + 2026-06-16T14:31:36Z + + False + + False + False + AAAAAAAAAAAAAAAAAAAAAA== + + 3zb/pabKWkmEeL/izyDVwg== + 0 + + + + + + 2026-06-16T15:15:24Z + 2026-06-16T15:15:50Z + 2026-06-16T15:15:55Z + 2026-06-03T13:08:10Z + False + 2 + 2026-06-16T15:15:55Z + + + Notes + + + + Password + WuNgQly6ZaLGX2BrwBCt + + + Title + Test secret test equipment + + + URL + + + + UserName + test.user2 + + + True + 0 + + + + + ++n1F/0kiUCOhAuWaEuhFw== + General + + 48 + + 2026-06-16T14:29:35Z + 2026-06-16T14:29:35Z + 2026-06-16T14:31:36Z + 2026-06-03T13:08:10Z + False + 2 + 2026-06-16T14:31:36Z + + True + + Null + Null + AAAAAAAAAAAAAAAAAAAAAA== + + + 4o0sdQBHfUucE/5/kTCrWg== + Windows + + 38 + + 2026-06-16T14:29:35Z + 2026-06-16T14:29:35Z + 2026-06-16T14:31:48Z + 2026-06-03T13:08:10Z + False + 2 + 2026-06-16T14:31:48Z + + True + + Null + Null + AAAAAAAAAAAAAAAAAAAAAA== + + + CdKlK4LLpUCDkeQ0ecfH0w== + Network + + 3 + + 2026-06-16T14:29:35Z + 2026-06-16T14:29:35Z + 2026-06-16T14:32:41Z + 2026-06-03T13:08:10Z + False + 2 + 2026-06-16T14:32:41Z + + True + + Null + Null + AAAAAAAAAAAAAAAAAAAAAA== + + + xIReP4+wx0exIIvtZjoYOw== + Internet + + 1 + + 2026-06-16T14:29:35Z + 2026-06-16T14:29:35Z + 2026-06-16T14:32:44Z + 2026-06-03T13:08:10Z + False + 2 + 2026-06-16T14:32:44Z + + True + + Null + Null + AAAAAAAAAAAAAAAAAAAAAA== + + + fn2Sea6Gyki2Wq68c3pDdQ== + eMail + + 19 + + 2026-06-16T14:29:35Z + 2026-06-16T14:29:35Z + 2026-06-16T14:32:48Z + 2026-06-03T13:08:10Z + False + 2 + 2026-06-16T14:32:48Z + + True + + Null + Null + AAAAAAAAAAAAAAAAAAAAAA== + + + 9E9FN7+tFEG0d9o4Gibflw== + Homebanking + + 37 + + 2026-06-16T14:29:35Z + 2026-06-16T14:29:35Z + 2026-06-16T14:32:52Z + 2026-06-03T13:08:10Z + False + 3 + 2026-06-16T14:32:52Z + + True + + Null + Null + AAAAAAAAAAAAAAAAAAAAAA== + + + + AVSR521hIU6aQgsBDpsLHw== + Digital Radio + + 48 + + 2026-06-16T14:31:51Z + 2026-06-16T14:31:59Z + 2026-06-16T15:18:56Z + 2026-06-15T23:00:00Z + False + 13 + 2026-06-16T14:31:51Z + + True + + Null + Null + AAAAAAAAAAAAAAAAAAAAAA== + + p/1clFGvQ0yC3QDR8MduEQ== + Mier + + 48 + + 2026-06-16T14:32:16Z + 2026-06-16T14:32:21Z + 2026-06-16T15:16:44Z + 2026-06-15T23:00:00Z + False + 5 + 2026-06-16T14:32:16Z + + True + + Null + Null + iPGUG7OkvUmTgbC86cl/3g== + + iPGUG7OkvUmTgbC86cl/3g== + 0 + + + + + + 2026-06-16T14:36:16Z + 2026-06-16T15:17:26Z + 2026-06-16T15:18:01Z + 2026-06-03T13:08:10Z + False + 6 + 2026-06-16T14:36:16Z + + + Notes + This is a test +Mier 1 + + + Password + Password4! + + + Title + Test Secret Mier 1 + + + URL + www.google.com + + + UserName + Test.user + + + True + 0 + + + + iPGUG7OkvUmTgbC86cl/3g== + 0 + + + + + + 2026-06-16T14:36:16Z + 2026-06-16T14:38:13Z + 2026-06-16T15:17:51Z + 2026-06-03T13:08:10Z + False + 3 + 2026-06-16T14:36:16Z + + + Notes + This is a test +Mier 1 + + + Password + Password1! + + + Title + Test Secret Mier 1 + + + URL + www.google.com + + + UserName + Test.user + + + True + 0 + + + + iPGUG7OkvUmTgbC86cl/3g== + 0 + + + + + + 2026-06-16T14:36:16Z + 2026-06-16T15:17:09Z + 2026-06-16T15:17:56Z + 2026-06-03T13:08:10Z + False + 4 + 2026-06-16T14:36:16Z + + + Notes + This is a test +Mier 1 + + + Password + Password2! + + + Title + Test Secret Mier 1 + + + URL + www.google.com + + + UserName + Test.user + + + True + 0 + + + + iPGUG7OkvUmTgbC86cl/3g== + 0 + + + + + + 2026-06-16T14:36:16Z + 2026-06-16T15:17:17Z + 2026-06-16T15:18:00Z + 2026-06-03T13:08:10Z + False + 5 + 2026-06-16T14:36:16Z + + + Notes + This is a test +Mier 1 + + + Password + Password3! + + + Title + Test Secret Mier 1 + + + URL + www.google.com + + + UserName + Test.user + + + True + 0 + + + + + + ZvJgr8KxAUC9Pd3RcxQOsQ== + 0 + + + + + + 2026-06-16T14:38:18Z + 2026-06-16T15:18:29Z + 2026-06-16T15:18:55Z + 2026-06-03T13:08:10Z + False + 5 + 2026-06-16T14:38:18Z + + + Notes + This is a test +Mier 2 + + + Password + Password4! + + + Title + Test Secret Mier 2 + + + URL + www.google.com + + + UserName + test.user2 + + + True + 0 + + + + ZvJgr8KxAUC9Pd3RcxQOsQ== + 0 + + + + + + 2026-06-16T14:38:18Z + 2026-06-16T14:39:02Z + 2026-06-16T14:39:02Z + 2026-06-03T13:08:10Z + False + 1 + 2026-06-16T14:38:18Z + + + Notes + This is a test +Mier 2 + + + Password + Password1! + + + Title + Test Secret Mier 2 + + + URL + www.google.com + + + UserName + test.user2 + + + True + 0 + + + + ZvJgr8KxAUC9Pd3RcxQOsQ== + 0 + + + + + + 2026-06-16T14:38:18Z + 2026-06-16T15:18:11Z + 2026-06-16T15:18:11Z + 2026-06-03T13:08:10Z + False + 2 + 2026-06-16T14:38:18Z + + + Notes + This is a test +Mier 2 + + + Password + Password2! + + + Title + Test Secret Mier 2 + + + URL + www.google.com + + + UserName + test.user2 + + + True + 0 + + + + ZvJgr8KxAUC9Pd3RcxQOsQ== + 0 + + + + + + 2026-06-16T14:38:18Z + 2026-06-16T15:18:20Z + 2026-06-16T15:18:20Z + 2026-06-03T13:08:10Z + False + 3 + 2026-06-16T14:38:18Z + + + Notes + This is a test +Mier 2 + + + Password + Password3! + + + Title + Test Secret Mier 2 + + + URL + www.google.com + + + UserName + test.user2 + + + True + 0 + + + + + + + 81XGZA2StUK0ca8hA5F9cw== + Baseband + + 48 + + 2026-06-16T14:33:04Z + 2026-06-16T14:33:09Z + 2026-06-16T15:18:56Z + 2026-06-15T23:00:00Z + False + 4 + 2026-06-16T14:33:04Z + + True + + Null + Null + GMS1T4gYGEq14TU1z8Lz8Q== + + GMS1T4gYGEq14TU1z8Lz8Q== + 0 + + + + + + 2026-06-16T14:39:37Z + 2026-06-16T15:19:05Z + 2026-06-16T15:19:05Z + 2026-06-03T13:08:10Z + False + 2 + 2026-06-16T14:39:37Z + + + Notes + This is a test +Baseband 1 + + + Password + Password2! + + + Title + Test Secret Baseband 1 + + + URL + + + + UserName + test.user + + + True + 0 + + + + GMS1T4gYGEq14TU1z8Lz8Q== + 0 + + + + + + 2026-06-16T14:39:37Z + 2026-06-16T14:40:29Z + 2026-06-16T14:40:29Z + 2026-06-03T13:08:10Z + False + 1 + 2026-06-16T14:39:37Z + + + Notes + This is a test +Baseband 1 + + + Password + Password1! + + + Title + Test Secret Baseband 1 + + + URL + + + + UserName + test.user + + + True + 0 + + + + + + + + OquroSdDdU+EmJLT+5DBqA== + Television + + 48 + + 2026-06-16T14:33:22Z + 2026-06-16T14:33:26Z + 2026-06-16T15:20:18Z + 2026-06-15T23:00:00Z + False + 23 + 2026-06-16T14:33:22Z + + True + + Null + Null + AAAAAAAAAAAAAAAAAAAAAA== + + GePeTGn5Gke7SBWvLmYCsA== + Transposers + + 48 + + 2026-06-16T14:33:46Z + 2026-06-16T14:33:52Z + 2026-06-16T15:20:18Z + 2026-06-15T23:00:00Z + False + 14 + 2026-06-16T14:33:46Z + + True + + Null + Null + AAAAAAAAAAAAAAAAAAAAAA== + + YmWvLaNgpUSIt5j5Ln30dg== + 0 + + + + + + 2026-06-16T14:40:51Z + 2026-06-16T15:19:29Z + 2026-06-16T15:19:29Z + 2026-06-03T13:08:10Z + False + 3 + 2026-06-16T14:40:51Z + + + Notes + This is a test +Transposer 1 + + + Password + Password3! + + + Title + Test secret transposer 1 + + + URL + www.google.com + + + UserName + test.user3 + + + True + 0 + + + + YmWvLaNgpUSIt5j5Ln30dg== + 0 + + + + + + 2026-06-16T14:40:51Z + 2026-06-16T15:03:20Z + 2026-06-16T15:03:20Z + 2026-06-03T13:08:10Z + False + 1 + 2026-06-16T14:40:51Z + + + Notes + This is a test +Transposer 1 + + + Password + Password1! + + + Title + Test secret transposer 1 + + + URL + www.google.com + + + UserName + test.user3 + + + True + 0 + + + + YmWvLaNgpUSIt5j5Ln30dg== + 0 + + + + + + 2026-06-16T14:40:51Z + 2026-06-16T15:19:20Z + 2026-06-16T15:19:20Z + 2026-06-03T13:08:10Z + False + 2 + 2026-06-16T14:40:51Z + + + Notes + This is a test +Transposer 1 + + + Password + Password2! + + + Title + Test secret transposer 1 + + + URL + www.google.com + + + UserName + test.user3 + + + True + 0 + + + + + + jTHFF4wiokCkzXG+SW1HZg== + R&S Relays + + 48 + + 2026-06-16T14:34:00Z + 2026-06-16T14:34:06Z + 2026-06-16T15:20:18Z + 2026-06-15T23:00:00Z + False + 6 + 2026-06-16T14:34:00Z + + True + + Null + Null + AAAAAAAAAAAAAAAAAAAAAA== + + vamcp8e54EaE4nSb6Uj4Jg== + 0 + + + + + + 2026-06-16T15:12:26Z + 2026-06-16T15:19:53Z + 2026-06-16T15:20:30Z + 2026-06-03T13:08:10Z + False + 4 + 2026-06-16T15:12:26Z + + + Notes + This is a test +Scret relay 1 + + + Password + Password3Password3 + + + Title + Test Secret R&S Relay + + + URL + + + + UserName + test.user + + + True + 0 + + + + vamcp8e54EaE4nSb6Uj4Jg== + 0 + + + + + + 2026-06-16T15:12:26Z + 2026-06-16T15:13:49Z + 2026-06-16T15:13:49Z + 2026-06-03T13:08:10Z + False + 1 + 2026-06-16T15:12:26Z + + + Notes + This is a test +Scret relay 1 + + + Password + Password1Password1 + + + Title + Test Secret R&S Relay + + + URL + + + + UserName + test.user + + + True + 0 + + + + vamcp8e54EaE4nSb6Uj4Jg== + 0 + + + + + + 2026-06-16T15:12:26Z + 2026-06-16T15:19:42Z + 2026-06-16T15:19:42Z + 2026-06-03T13:08:10Z + False + 2 + 2026-06-16T15:12:26Z + + + Notes + This is a test +Scret relay 1 + + + Password + Password2Password2 + + + Title + Test Secret R&S Relay + + + URL + + + + UserName + test.user + + + True + 0 + + + + + + + + s5P2xgRaMk2ovXD+Ge2JWQ== + NEC Transmitters + + 48 + + 2026-06-16T14:34:18Z + 2026-06-16T14:34:24Z + 2026-06-16T15:20:04Z + 2026-06-15T23:00:00Z + False + 5 + 2026-06-16T14:34:18Z + + True + + Null + Null + AAAAAAAAAAAAAAAAAAAAAA== + + XamnkswxUkGrq4SUDPRg6A== + 0 + + + + + + 2026-06-16T15:14:01Z + 2026-06-16T15:15:18Z + 2026-06-16T15:15:18Z + 2026-06-03T13:08:10Z + False + 1 + 2026-06-16T15:14:01Z + + + Notes + + + + Password + Password1! + + + Title + Test secret NEC Transmitters + + + URL + + + + UserName + Test.user + + + True + 0 + + + + + + + jN/K/nSRM0qNnq7W20TtTg== + Test Equipment + + 48 + + 2026-06-16T14:34:58Z + 2026-06-16T14:35:05Z + 2026-06-16T15:20:05Z + 2026-06-15T23:00:00Z + False + 3 + 2026-06-16T14:34:58Z + + True + + Null + Null + P5SESXNgZkmTi6cyhkKReg== + + P5SESXNgZkmTi6cyhkKReg== + 0 + + + + + + 2026-06-16T15:15:58Z + 2026-06-16T15:20:13Z + 2026-06-16T15:20:13Z + 2026-06-03T13:08:10Z + False + 2 + 2026-06-16T15:15:58Z + + + Notes + This is a test +test equipment + + + Password + Password2! + + + Title + Test secret test equipment + + + URL + + + + UserName + test.user2 + + + True + 0 + + + + P5SESXNgZkmTi6cyhkKReg== + 0 + + + + + + 2026-06-16T15:15:58Z + 2026-06-16T15:16:35Z + 2026-06-16T15:16:35Z + 2026-06-03T13:08:10Z + False + 1 + 2026-06-16T15:15:58Z + + + Notes + This is a test +test equipment + + + Password + Password1! + + + Title + Test secret test equipment + + + URL + + + + UserName + test.user2 + + + True + 0 + + + + + + + + + \ No newline at end of file diff --git a/1password/migration/keepass/import-from-keepass-xml.py b/1password/migration/keepass/import-from-keepass-xml.py index 153baca..2c52fa5 100644 --- a/1password/migration/keepass/import-from-keepass-xml.py +++ b/1password/migration/keepass/import-from-keepass-xml.py @@ -3,25 +3,26 @@ KeePass XML → 1Password migration helper — SDK bulk create Reads a KeePass XML export (.xml, produced via File → Export → KeePass XML 2.x -in KeePass or compatible apps) and imports every entry into 1Password using the -onepassword-sdk bulk-create API. +in KeePass or compatible apps) and imports every entry into a single 1Password +vault using the onepassword-sdk bulk-create API. -PASSWORD HISTORY - Each entry's child entries are collected, deduplicated, and appended - to the item's Notes field as a human-readable block: - - --- Password History --- - 2024-01-15T09:23:00Z hunter2 - 2023-06-01T14:00:00Z correct-horse +The vault is created by the service account if it does not already exist. - The most-recent value in that differs from the current password is - listed first (newest → oldest). Duplicate passwords are suppressed. +FOLDER → TAG MAPPING + KeePass group paths are converted to tags on each item: + - Top-level group "Email" → tag "Email" + - Nested group "Email/Work" → tags "Email" and "Email/Work" + Items at the database root (no group) receive no tags. -FOLDER → VAULT MAPPING - Without --collapse-folders: - Each KeePass group path (e.g. "Email/Work") becomes its own vault. - With --collapse-folders: - All items land in the top-level group vault; sub-group names become tags. +PASSWORD HISTORY + KeePass password history is replayed as real 1Password item history: + 1. The item is created with the oldest historical password. + 2. The item is updated (put) once per subsequent historical password, + oldest → newest, ending with the current password. + Each put() call produces a genuine 1Password history entry, visible in the + UI via "View password history". Items with no history are bulk-created + normally. Dates are not preserved (1Password history timestamps reflect + when the import ran). ATTACHMENTS Binary attachments stored in elements are imported as 1Password @@ -34,17 +35,13 @@ REQUIREMENTS - OP_SERVICE_ACCOUNT_TOKEN env var - onepassword-sdk (auto-installed into .venv-1pw if missing) - - `op` CLI in PATH (only needed for user vault permission grants) USAGE python import-from-keepass-xml.py \\ --input keepass-export.xml \\ - --employee-vault "KeePass Import" \\ - [--private-prefix "Private - "] \\ - [--collapse-folders] \\ + --vault "KeePass Import" \\ [--dry-run] \\ - [--silent] \\ - [--user-for-private user@example.com] + [--silent] """ from __future__ import annotations @@ -56,13 +53,9 @@ import json import os import re -import subprocess import sys import xml.etree.ElementTree as ET -from collections import defaultdict from dataclasses import dataclass, field -from pathlib import Path -from shutil import which from typing import Dict, List, Optional, Set, Tuple # --------------------------------------------------------------------------- @@ -73,25 +66,12 @@ from onepassword import ( AutofillBehavior, FileCreateParams, - GroupAccess, ItemCategory, ItemCreateParams, ItemField, ItemFieldType, ItemSection, ItemsUpdateAllResponse, - READ_ITEMS, - REVEAL_ITEM_PASSWORD, - UPDATE_ITEM_HISTORY, - CREATE_ITEMS, - UPDATE_ITEMS, - ARCHIVE_ITEMS, - DELETE_ITEMS, - IMPORT_ITEMS, - EXPORT_ITEMS, - SEND_ITEMS, - PRINT_ITEMS, - MANAGE_VAULT, VaultCreateParams, VaultListParams, Website, @@ -105,7 +85,10 @@ _venv_python = os.path.join(_venv_dir, "bin", "python") if sys.executable == _venv_python: - print("ERROR: onepassword-sdk failed to import even inside the venv.", file=sys.stderr) + print( + "ERROR: onepassword-sdk failed to import even inside the venv.", + file=sys.stderr, + ) sys.exit(1) if not os.path.isfile(_venv_python): @@ -115,10 +98,14 @@ _req_file = os.path.join(_script_dir, "requirements.txt") if os.path.isfile(_req_file): print(f"Installing packages from {_req_file}...") - _sp.check_call([_venv_python, "-m", "pip", "install", "--quiet", "-r", _req_file]) + _sp.check_call( + [_venv_python, "-m", "pip", "install", "--quiet", "-r", _req_file] + ) else: print("Installing onepassword-sdk...") - _sp.check_call([_venv_python, "-m", "pip", "install", "--quiet", "onepassword-sdk"]) + _sp.check_call( + [_venv_python, "-m", "pip", "install", "--quiet", "onepassword-sdk"] + ) print("Restarting inside virtual environment...\n") os.execv(_venv_python, [_venv_python] + sys.argv) @@ -128,6 +115,7 @@ # 1Password client helpers (identical pattern to the Keeper migration script) # --------------------------------------------------------------------------- + async def _get_client() -> Client: token = os.getenv("OP_SERVICE_ACCOUNT_TOKEN") if not token: @@ -202,69 +190,16 @@ async def _ensure_vault( return created.id -# Permission bitmasks -_PERM_VIEWING = READ_ITEMS | REVEAL_ITEM_PASSWORD | UPDATE_ITEM_HISTORY -_PERM_EDITING = _PERM_VIEWING | CREATE_ITEMS | UPDATE_ITEMS | ARCHIVE_ITEMS | DELETE_ITEMS | IMPORT_ITEMS | EXPORT_ITEMS | SEND_ITEMS | PRINT_ITEMS -_PERM_MANAGING = _PERM_EDITING | MANAGE_VAULT - - -def _perms_list(manage_users: bool, manage_records: bool) -> List[str]: - perms = ["allow_viewing"] - if manage_records: - perms.append("allow_editing") - if manage_users: - perms.append("allow_managing") - return perms - - -def _run_op(cmd: List[str]) -> subprocess.CompletedProcess: - return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) - - -def _op_exists() -> bool: - return which("op") is not None - - -def _grant_user_permissions( - vault_name: str, - user_name: str, - *, - manage_users: bool, - manage_records: bool, - dry: bool, - silent: bool, -) -> None: - perms = _perms_list(manage_users, manage_records) - if dry: - if not silent: - print(f"DRY-RUN: would grant user {user_name!r} on {vault_name!r}: {perms}") - return - if not _op_exists(): - print( - f"WARN: 'op' CLI not found — cannot grant user {user_name!r} on {vault_name!r}. " - f"Run manually: op vault user grant --vault {vault_name!r} --user {user_name!r} " - f"--permissions {','.join(perms)}", - file=sys.stderr, - ) - return - cmd = ["op", "vault", "user", "grant", "--no-input", - "--vault", vault_name, "--user", user_name, - "--permissions", ",".join(perms)] - proc = _run_op(cmd) - if proc.returncode != 0: - print(f"WARN granting user {user_name!r} on {vault_name!r}: {proc.stderr.strip()}", file=sys.stderr) - elif not silent: - print(f"✔ Granted user {user_name!r} on vault {vault_name!r}: {perms}") - - # --------------------------------------------------------------------------- # Data models # --------------------------------------------------------------------------- + @dataclass class PasswordHistoryEntry: """A single historical password with its last-modification timestamp.""" - timestamp: str # ISO-8601 string from KeePass + + timestamp: str # ISO-8601 string from KeePass password: str @@ -282,7 +217,7 @@ class Record: login_url: Optional[str] notes: Optional[str] otpauth: Optional[str] - group_path: List[str] # e.g. ["Email", "Work"] + group_path: List[str] # e.g. ["Email", "Work"] attachments: List[InMemoryAttachment] = field(default_factory=list) password_history: List[PasswordHistoryEntry] = field(default_factory=list) @@ -291,6 +226,7 @@ class Record: # Resumable state file # --------------------------------------------------------------------------- + def _item_fingerprint(vault_id: str, rec: Record) -> str: parts = [ vault_id, @@ -323,7 +259,10 @@ def load_state(input_path: str, *, silent: bool) -> Set[str]: with open(path, "r", encoding="utf-8") as f: data = json.load(f) except (json.JSONDecodeError, OSError) as e: - print(f"WARN: Could not read state file {path}: {e}. Starting fresh.", file=sys.stderr) + print( + f"WARN: Could not read state file {path}: {e}. Starting fresh.", + file=sys.stderr, + ) return set() fingerprints = data.get("completed", []) if _compute_checksum(fingerprints) != data.get("checksum", ""): @@ -337,7 +276,11 @@ def load_state(input_path: str, *, silent: bool) -> Set[str]: def save_state(input_path: str, completed: Set[str], *, silent: bool) -> None: path = _state_file_path(input_path) fingerprints = sorted(completed) - data = {"checksum": _compute_checksum(fingerprints), "completed": fingerprints, "count": len(fingerprints)} + data = { + "checksum": _compute_checksum(fingerprints), + "completed": fingerprints, + "count": len(fingerprints), + } tmp_path = path + ".tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) @@ -363,6 +306,7 @@ def _is_rate_limit_error(exc: Exception) -> bool: # KeePass XML parser # --------------------------------------------------------------------------- + def _text(el: Optional[ET.Element], tag: str, default: str = "") -> str: """Return stripped text of a direct child element, or default.""" if el is None: @@ -387,6 +331,7 @@ def _find_string_value(entry_el: ET.Element, key: str) -> Optional[str]: _OTPAUTH_RE = re.compile(r"otpauth://[^\s]+") + def _extract_otpauth(entry_el: ET.Element) -> Optional[str]: """Look for an otpauth:// URI in any String value or the Notes field.""" for s in entry_el.findall("String"): @@ -398,7 +343,9 @@ def _extract_otpauth(entry_el: ET.Element) -> Optional[str]: return None -def _parse_history(entry_el: ET.Element, current_password: Optional[str]) -> List[PasswordHistoryEntry]: +def _parse_history( + entry_el: ET.Element, current_password: Optional[str] +) -> List[PasswordHistoryEntry]: """ Extract … blocks. @@ -445,7 +392,9 @@ def _format_history_block(history: List[PasswordHistoryEntry]) -> str: return "\n".join(lines) -def _parse_attachments(entry_el: ET.Element, binary_pool: Dict[str, bytes]) -> List[InMemoryAttachment]: +def _parse_attachments( + entry_el: ET.Element, binary_pool: Dict[str, bytes] +) -> List[InMemoryAttachment]: """ KeePass XML stores attachment content in a global pool, referenced by Ref id from . @@ -463,7 +412,10 @@ def _parse_attachments(entry_el: ET.Element, binary_pool: Dict[str, bytes]) -> L if ref is not None: content = binary_pool.get(ref) if content is None: - print(f"WARN: attachment ref {ref!r} ('{name}') not found in binary pool", file=sys.stderr) + print( + f"WARN: attachment ref {ref!r} ('{name}') not found in binary pool", + file=sys.stderr, + ) continue else: # Inline base64 @@ -473,7 +425,9 @@ def _parse_attachments(entry_el: ET.Element, binary_pool: Dict[str, bytes]) -> L try: content = base64.b64decode(raw) except Exception as e: - print(f"WARN: could not decode attachment '{name}': {e}", file=sys.stderr) + print( + f"WARN: could not decode attachment '{name}': {e}", file=sys.stderr + ) continue attachments.append(InMemoryAttachment(name=name, content=content)) return attachments @@ -504,10 +458,14 @@ def _build_binary_pool(root: ET.Element) -> Dict[str, bytes]: data = base64.b64decode(raw) if compressed: import zlib + data = zlib.decompress(data, wbits=-15) # raw deflate pool[ref_id] = data except Exception as e: - print(f"WARN: could not decode binary pool entry id={ref_id}: {e}", file=sys.stderr) + print( + f"WARN: could not decode binary pool entry id={ref_id}: {e}", + file=sys.stderr, + ) return pool @@ -536,12 +494,12 @@ def _walk_group( if entry_el.find("History") is None and entry_el.getparent() is not None: pass # ElementTree doesn't expose getparent; this is always fine - title = _find_string_value(entry_el, "Title") or "Untitled" - login = _find_string_value(entry_el, "UserName") or None + title = _find_string_value(entry_el, "Title") or "Untitled" + login = _find_string_value(entry_el, "UserName") or None password = _find_string_value(entry_el, "Password") or None - url = _find_string_value(entry_el, "URL") or None - notes = _find_string_value(entry_el, "Notes") or None - otpauth = _extract_otpauth(entry_el) + url = _find_string_value(entry_el, "URL") or None + notes = _find_string_value(entry_el, "Notes") or None + otpauth = _extract_otpauth(entry_el) # Suppress the otpauth string from appearing raw in notes if notes and otpauth and otpauth in notes: @@ -550,21 +508,29 @@ def _walk_group( history = _parse_history(entry_el, password) attachments = _parse_attachments(entry_el, binary_pool) - records.append(Record( - title=title, - login=login, - password=password, - login_url=url, - notes=notes, - otpauth=otpauth, - group_path=current_path, - attachments=attachments, - password_history=history, - )) + records.append( + Record( + title=title, + login=login, + password=password, + login_url=url, + notes=notes, + otpauth=otpauth, + group_path=current_path, + attachments=attachments, + password_history=history, + ) + ) # Recurse into sub-groups for sub_group in group_el.findall("Group"): - _walk_group(sub_group, current_path, binary_pool, records, skip_recycle_bin=skip_recycle_bin) + _walk_group( + sub_group, + current_path, + binary_pool, + records, + skip_recycle_bin=skip_recycle_bin, + ) def load_keepass_xml(input_path: str) -> List[Record]: @@ -577,8 +543,11 @@ def load_keepass_xml(input_path: str) -> List[Record]: root = tree.getroot() if root.tag != "KeePassFile": - print(f"ERROR: Expected root tag , got <{root.tag}>. " - "Is this a KeePass XML 2.x export?", file=sys.stderr) + print( + f"ERROR: Expected root tag , got <{root.tag}>. " + "Is this a KeePass XML 2.x export?", + file=sys.stderr, + ) sys.exit(2) binary_pool = _build_binary_pool(root) @@ -596,22 +565,29 @@ def load_keepass_xml(input_path: str) -> List[Record]: # The outermost group is the database name — skip it from path labelling. for entry_el in root_group.findall("Entry"): - title = _find_string_value(entry_el, "Title") or "Untitled" - login = _find_string_value(entry_el, "UserName") or None + title = _find_string_value(entry_el, "Title") or "Untitled" + login = _find_string_value(entry_el, "UserName") or None password = _find_string_value(entry_el, "Password") or None - url = _find_string_value(entry_el, "URL") or None - notes = _find_string_value(entry_el, "Notes") or None - otpauth = _extract_otpauth(entry_el) + url = _find_string_value(entry_el, "URL") or None + notes = _find_string_value(entry_el, "Notes") or None + otpauth = _extract_otpauth(entry_el) if notes and otpauth and otpauth in notes: notes = notes.replace(otpauth, "").strip() or None history = _parse_history(entry_el, password) attachments = _parse_attachments(entry_el, binary_pool) - records.append(Record( - title=title, login=login, password=password, - login_url=url, notes=notes, otpauth=otpauth, - group_path=[], # top-level items have no group path - attachments=attachments, password_history=history, - )) + records.append( + Record( + title=title, + login=login, + password=password, + login_url=url, + notes=notes, + otpauth=otpauth, + group_path=[], # top-level items have no group path + attachments=attachments, + password_history=history, + ) + ) for sub_group in root_group.findall("Group"): _walk_group(sub_group, [], binary_pool, records) @@ -620,49 +596,34 @@ def load_keepass_xml(input_path: str) -> List[Record]: # --------------------------------------------------------------------------- -# Notes field builder — merges original notes + history block +# Notes field builder # --------------------------------------------------------------------------- + def _build_notes(rec: Record) -> Optional[str]: - parts: List[str] = [] - if rec.notes: - parts.append(rec.notes) - if rec.password_history: - parts.append(_format_history_block(rec.password_history)) - return "\n\n".join(parts) if parts else None + """Return the item's original notes, unmodified. History is written via put().""" + return rec.notes or None # --------------------------------------------------------------------------- -# Vault name derivation from group path +# Tag derivation from group path # --------------------------------------------------------------------------- -def _path_to_vault_and_tags( - group_path: List[str], - *, - prefix: str, - collapse_folders: bool, -) -> Tuple[str, List[str]]: - """ - Map a group_path list to (vault_name, tags). - Without --collapse-folders: vault = prefix + full/path; no tags. - With --collapse-folders: vault = prefix + top-level group; - tag = "Parent\\Child\\Deeper" for sub-paths. +def _group_path_to_tags(group_path: List[str]) -> List[str]: """ - if not group_path: - return "", [] # caller uses employee_vault as fallback - - if not collapse_folders: - vault_name = prefix + "/".join(group_path) - return vault_name, [] + Convert a KeePass group path to a list of 1Password tags. - # collapse: parent vault only - vault_name = prefix + group_path[0] - if len(group_path) == 1: - tags = [group_path[0]] - else: - tags = ["\\".join(group_path)] - return vault_name, tags + Each level of the hierarchy gets its own tag so items are findable + by either the top-level group or the full path: + ["Email"] → ["Email"] + ["Email", "Work"] → ["Email", "Email/Work"] + Items at the database root (empty path) receive no tags. + """ + tags: List[str] = [] + for i in range(len(group_path)): + tags.append("/".join(group_path[: i + 1])) + return tags # --------------------------------------------------------------------------- @@ -671,8 +632,9 @@ def _path_to_vault_and_tags( BULK_CREATE_MAX = 100 + def _chunked(items: List, size: int) -> List[List]: - return [items[i: i + size] for i in range(0, len(items), size)] + return [items[i : i + size] for i in range(0, len(items), size)] def _make_file_params( @@ -698,23 +660,56 @@ def _build_login_params( rec: Record, notes: Optional[str], tags: Optional[List[str]], + *, + password_override: Optional[str] = None, ) -> ItemCreateParams: fields: List[ItemField] = [] + password_value = ( + password_override if password_override is not None else rec.password + ) if rec.login is not None: - fields.append(ItemField(id="username", value=rec.login, title="Username", fieldType=ItemFieldType.TEXT)) - if rec.password is not None: - fields.append(ItemField(id="password", value=rec.password, title="Password", fieldType=ItemFieldType.CONCEALED)) + fields.append( + ItemField( + id="username", + value=rec.login, + title="Username", + fieldType=ItemFieldType.TEXT, + ) + ) + if password_value is not None: + fields.append( + ItemField( + id="password", + value=password_value, + title="Password", + fieldType=ItemFieldType.CONCEALED, + ) + ) sections: List[ItemSection] = [] if rec.otpauth: sections.append(ItemSection(id="sec-otp", title="Two-Factor")) - fields.append(ItemField(id="otp", title="OTP", fieldType=ItemFieldType.TOTP, value=rec.otpauth, section_id="sec-otp")) + fields.append( + ItemField( + id="otp", + title="OTP", + fieldType=ItemFieldType.TOTP, + value=rec.otpauth, + section_id="sec-otp", + ) + ) files = _make_file_params(rec.attachments, sections) websites: List[Website] = [] if rec.login_url: - websites.append(Website(url=rec.login_url, label="site", autofill_behavior=AutofillBehavior.ANYWHEREONWEBSITE)) + websites.append( + Website( + url=rec.login_url, + label="site", + autofill_behavior=AutofillBehavior.ANYWHEREONWEBSITE, + ) + ) return ItemCreateParams( title=rec.title, @@ -741,14 +736,46 @@ def _build_secure_note_params( if rec.login or rec.password or rec.login_url: sections.append(ItemSection(id="details", title="Details")) if rec.login: - fields.append(ItemField(id="username", value=rec.login, title="Username", fieldType=ItemFieldType.TEXT, section_id="details")) + fields.append( + ItemField( + id="username", + value=rec.login, + title="Username", + fieldType=ItemFieldType.TEXT, + section_id="details", + ) + ) if rec.password: - fields.append(ItemField(id="password", value=rec.password, title="Password", fieldType=ItemFieldType.CONCEALED, section_id="details")) + fields.append( + ItemField( + id="password", + value=rec.password, + title="Password", + fieldType=ItemFieldType.CONCEALED, + section_id="details", + ) + ) if rec.login_url: - fields.append(ItemField(id="url", value=rec.login_url, title="URL", fieldType=ItemFieldType.TEXT, section_id="details")) + fields.append( + ItemField( + id="url", + value=rec.login_url, + title="URL", + fieldType=ItemFieldType.TEXT, + section_id="details", + ) + ) if rec.otpauth: sections.append(ItemSection(id="sec-otp", title="Two-Factor")) - fields.append(ItemField(id="otp", title="OTP", fieldType=ItemFieldType.TOTP, value=rec.otpauth, section_id="sec-otp")) + fields.append( + ItemField( + id="otp", + title="OTP", + fieldType=ItemFieldType.TOTP, + value=rec.otpauth, + section_id="sec-otp", + ) + ) files = _make_file_params(rec.attachments, sections) @@ -777,6 +804,7 @@ def _categorize(rec: Record) -> str: # Planner + bulk create # --------------------------------------------------------------------------- + @dataclass class _PendingItem: params: ItemCreateParams @@ -784,61 +812,86 @@ class _PendingItem: rec: Record +def _set_password_on_item(item, password: str) -> None: + """Mutate a live 1Password item object's password field value in place.""" + for f in item.fields or []: + if getattr(f, "id", None) == "password": + f.value = password + return + # Field not found — nothing to mutate (item has no password field) + + +async def _replay_history(client, item, rec: Record, *, silent: bool) -> bool: + """ + Replay KeePass password history as real 1Password item history. + + Each put() causes 1Password to save the *previous* live value into history. + To produce history [A, B, C] with current password D (D not in history): + + create(A) → put(B) → put(C) → put(D) + saves A saves B saves C + to hist to hist to hist + + The item is already created with the oldest password (A) by the caller, + so we just put() each subsequent value in order, ending with the current. + + Returns True on success, False if rate-limited. + """ + if not rec.password_history: + return True + + # history is newest-first; reverse to oldest-first, skip the first entry + # (oldest) because the item was already created with that value. + oldest_first = list(reversed(rec.password_history)) + + for h in oldest_first[1:]: + _set_password_on_item(item, h.password) + try: + item = await client.items.put(item) + except Exception as e: + if _is_rate_limit_error(e): + return False + print( + f"WARN: failed to write history entry for '{rec.title}': {e}", + file=sys.stderr, + ) + return True # non-fatal: skip remaining history for this item + + # Final put() sets the current password as the live value. + # This saves the last historical password (C) into history, not D. + _set_password_on_item(item, rec.password or "") + try: + await client.items.put(item) + except Exception as e: + if _is_rate_limit_error(e): + return False + print( + f"WARN: failed to restore current password for '{rec.title}': {e}", + file=sys.stderr, + ) + return True + + async def plan_and_apply( records: List[Record], *, input_path: str, - employee_vault: str, - private_prefix: str, - collapse_folders: bool, + vault_name: str, dry: bool, silent: bool, - user_for_private: Optional[str], ) -> None: client = await _get_client() completed = load_state(input_path, silent=silent) if not dry else set() - # Collect all vault names we'll need - vault_set: Set[str] = {employee_vault} - for rec in records: - vn, _ = _path_to_vault_and_tags(rec.group_path, prefix=private_prefix, collapse_folders=collapse_folders) - if vn: - vault_set.add(vn) - - if not silent: - print(f"Using {len(vault_set)} vault(s): {', '.join(sorted(vault_set))}") - - # Ensure all vaults exist - name_to_id, vault_titles = await _vault_name_to_id_map(client) - need_create = [v for v in vault_set if v not in name_to_id and _normalize_vault_name(v) not in name_to_id] - if need_create and not silent: - print(f"Vault(s) to create: {', '.join(sorted(need_create))}") - for v in sorted(need_create): - await _ensure_vault(client, v, name_to_id, dry=dry, silent=silent) - if not dry and need_create: - name_to_id, vault_titles = await _vault_name_to_id_map(client) - - resolved: Dict[str, str] = {} - for v in vault_set: - if dry and v in need_create: - resolved[v] = "" - else: - resolved[v] = _resolve_vault_id(name_to_id, v) - - # Optional user grants on private vaults - if user_for_private: - for vn in vault_set: - _grant_user_permissions( - vn, user_for_private, - manage_users=(vn == employee_vault), - manage_records=True, - dry=dry, silent=silent, - ) + # Ensure the single destination vault exists (create if needed) + name_to_id, _ = await _vault_name_to_id_map(client) + vault_id = await _ensure_vault( + client, vault_name, name_to_id, dry=dry, silent=silent + ) if dry: for rec in records: - vn, tags = _path_to_vault_and_tags(rec.group_path, prefix=private_prefix, collapse_folders=collapse_folders) - vault_name = vn or employee_vault + tags = _group_path_to_tags(rec.group_path) category = _categorize(rec) hist_count = len(rec.password_history) att_names = [a.name for a in rec.attachments] @@ -848,40 +901,52 @@ async def plan_and_apply( if rec.otpauth: msg += " +TOTP" if hist_count: - msg += f" +{hist_count} history entries" + msg += f" +{hist_count} history put() calls" if att_names: msg += f" +files {att_names}" print(msg) return - # Build batches per vault_id - batches: Dict[str, List[_PendingItem]] = defaultdict(list) + # Split records into two groups: + # - no_history: eligible for bulk create_all (fast path) + # - with_history: must be created individually then put() N times + no_history: List[_PendingItem] = [] + with_history: List[_PendingItem] = [] skipped = 0 for rec in records: - vn, tags = _path_to_vault_and_tags(rec.group_path, prefix=private_prefix, collapse_folders=collapse_folders) - vault_name = vn or employee_vault - vault_id = resolved[vault_name] fp = _item_fingerprint(vault_id, rec) - if fp in completed: skipped += 1 continue + tags = _group_path_to_tags(rec.group_path) or None notes = _build_notes(rec) category = _categorize(rec) - if category == "Login": - params = _build_login_params(vault_id, rec, notes, tags or None) + if rec.password_history: + # Create with the oldest historical password (A). Each subsequent + # put() then shifts the window: put(B) saves A, put(C) saves B, + # put(D) saves C — so history ends up as [A, B, C] and D is live. + oldest_password = rec.password_history[-1].password # list is newest-first + if category == "Login": + params = _build_login_params( + vault_id, rec, notes, tags, password_override=oldest_password + ) + else: + params = _build_secure_note_params(vault_id, rec, notes, tags) + with_history.append(_PendingItem(params=params, fingerprint=fp, rec=rec)) else: - params = _build_secure_note_params(vault_id, rec, notes, tags or None) - - batches[vault_id].append(_PendingItem(params=params, fingerprint=fp, rec=rec)) + if category == "Login": + params = _build_login_params(vault_id, rec, notes, tags) + else: + params = _build_secure_note_params(vault_id, rec, notes, tags) + no_history.append(_PendingItem(params=params, fingerprint=fp, rec=rec)) if skipped and not silent: print(f"⏭ Skipping {skipped} already-completed items") - total_remaining = sum(len(v) for v in batches.values()) + total_remaining = len(no_history) + len(with_history) if total_remaining == 0: if not silent: print("✔ All items already imported — nothing to do") @@ -889,54 +954,94 @@ async def plan_and_apply( return if not silent: - print(f"📦 {total_remaining} items to create") + print( + f"📦 {total_remaining} items to create in vault '{vault_name}'" + + (f" ({len(with_history)} with history replay)" if with_history else "") + ) rate_limited = False - id_to_name = {vid: name for name, vid in resolved.items()} + total_ok = 0 - for vault_id, pending_list in batches.items(): - if not pending_list or rate_limited: + # ── Fast path: bulk create items with no history ────────────────────────── + for chunk in _chunked(no_history, BULK_CREATE_MAX): + if rate_limited: + break + try: + resp: ItemsUpdateAllResponse = await client.items.create_all( + vault_id, [item.params for item in chunk] + ) + except Exception as e: + if _is_rate_limit_error(e): + print( + f"\n⚠ Rate limited during bulk create. Saving progress...", + file=sys.stderr, + ) + rate_limited = True + break + print(f"ERROR bulk create in vault '{vault_name}': {e}", file=sys.stderr) continue - vault_title = id_to_name.get(vault_id, vault_id) - if not silent: - print(f"Creating {len(pending_list)} item(s) in vault '{vault_title}'...") - total_ok = 0 - for chunk in _chunked(pending_list, BULK_CREATE_MAX): - try: - resp: ItemsUpdateAllResponse = await client.items.create_all( - vault_id, [item.params for item in chunk] - ) - except Exception as e: - if _is_rate_limit_error(e): - print(f"\n⚠ Rate limited in vault '{vault_title}'. Saving progress...", file=sys.stderr) + for i, ir in enumerate(resp.individual_responses): + if ir.error is not None: + err_str = str(ir.error).lower() + if "429" in err_str or "rate limit" in err_str: + print( + f"\n⚠ Rate limited on item '{chunk[i].rec.title}'. Saving progress...", + file=sys.stderr, + ) rate_limited = True break - print(f"ERROR bulk create in vault {vault_title}: {e}", file=sys.stderr) - continue - - for i, ir in enumerate(resp.individual_responses): - if ir.error is not None: - err_str = str(ir.error).lower() - if "429" in err_str or "rate limit" in err_str: - print(f"\n⚠ Rate limited on item '{chunk[i].rec.title}'. Saving progress...", file=sys.stderr) - rate_limited = True - break - title = chunk[i].params.title if i < len(chunk) else "?" - print(f"ERROR creating '{title}': {ir.error}", file=sys.stderr) - else: - completed.add(chunk[i].fingerprint) - total_ok += 1 - - if rate_limited: + print( + f"ERROR creating '{chunk[i].params.title}': {ir.error}", + file=sys.stderr, + ) + else: + completed.add(chunk[i].fingerprint) + total_ok += 1 + + # ── Slow path: create + put() for items that have history ───────────────── + for pending in with_history: + if rate_limited: + break + try: + item = await client.items.create(pending.params) + except Exception as e: + if _is_rate_limit_error(e): + print( + f"\n⚠ Rate limited creating '{pending.rec.title}'. Saving progress...", + file=sys.stderr, + ) + rate_limited = True break + print(f"ERROR creating '{pending.rec.title}': {e}", file=sys.stderr) + continue + ok = await _replay_history(client, item, pending.rec, silent=silent) + if not ok: + print( + f"\n⚠ Rate limited replaying history for '{pending.rec.title}'. Saving progress...", + file=sys.stderr, + ) + rate_limited = True + break + + completed.add(pending.fingerprint) + total_ok += 1 if not silent: - print(f"✔ Bulk created {total_ok}/{len(pending_list)} items in vault '{vault_title}'") + hist_count = len(pending.rec.password_history) + print( + f" ✔ '{pending.rec.title}' created with {hist_count} history entries" + ) + + if not silent: + print(f"✔ Created {total_ok}/{total_remaining} items in vault '{vault_name}'") if rate_limited: save_state(input_path, completed, silent=silent) - print("\n🔄 Import paused due to rate limiting. Re-run the same command to resume.", file=sys.stderr) + print( + "\n🔄 Import paused due to rate limiting. Re-run the same command to resume.", + file=sys.stderr, + ) sys.exit(3) else: delete_state(input_path, silent=silent) @@ -948,18 +1053,25 @@ async def plan_and_apply( # Entrypoint # --------------------------------------------------------------------------- + async def main() -> None: ap = argparse.ArgumentParser( - description="KeePass XML → 1Password migration (SDK bulk create, password history in notes)" + description="KeePass XML → 1Password migration (single vault, group paths as tags)" + ) + ap.add_argument( + "--input", required=True, help="Path to KeePass XML 2.x export (.xml)" + ) + ap.add_argument( + "--vault", + required=True, + help="Destination vault name (created if it does not exist)", + ) + ap.add_argument( + "--dry-run", + action="store_true", + help="Print planned actions without creating anything", ) - ap.add_argument("--input", required=True, help="Path to KeePass XML 2.x export (.xml)") - ap.add_argument("--employee-vault", required=True, help="Fallback vault for items without a group") - ap.add_argument("--private-prefix", default="", help="Prefix for vault names derived from groups (default: none)") - ap.add_argument("--collapse-folders", action="store_true", - help="Collapse sub-groups into the parent vault; sub-groups become tags") - ap.add_argument("--dry-run", action="store_true", help="Print planned actions without creating anything") ap.add_argument("--silent", action="store_true", help="Suppress progress output") - ap.add_argument("--user-for-private", help="Grant this user (email) edit access to all vaults") args = ap.parse_args() @@ -979,23 +1091,24 @@ async def main() -> None: sys.exit(2) if not args.silent: - att_count = sum(len(r.attachments) for r in records) + att_count = sum(len(r.attachments) for r in records) hist_count = sum(len(r.password_history) for r in records) print( f"Loaded {len(records)} entries" + (f" ({att_count} attachments)" if att_count else "") - + (f" ({hist_count} historical passwords across all entries)" if hist_count else "") + + ( + f" ({hist_count} historical passwords across all entries)" + if hist_count + else "" + ) ) await plan_and_apply( records, input_path=os.path.abspath(args.input), - employee_vault=args.employee_vault, - private_prefix=args.private_prefix, - collapse_folders=args.collapse_folders, + vault_name=args.vault, dry=args.dry_run, silent=args.silent, - user_for_private=args.user_for_private, )