diff --git a/1password/migration/PPS/README.md b/1password/migration/PPS/README.md new file mode 100644 index 0000000..b37d675 --- /dev/null +++ b/1password/migration/PPS/README.md @@ -0,0 +1,362 @@ +# Pleasant Password Server → 1Password importer + +A single script, `keepass_to_1password.py`, that reads a Pleasant Password +Server XML export and imports it directly into 1Password using the official +[`onepassword-sdk`](https://pypi.org/project/onepassword-sdk/) Python SDK. + +Pleasant Password Server exports use the KeePass XML 2.x file format; the +script parses that structure as-is. + +## What it does + +1. **Parses** the XML export (`...`). +2. **One vault per subfolder.** Every folder `Group` that directly contains + entries becomes one 1Password vault. The vault title is + `" - "` — e.g. an export path of + `.../JAG/Client 1` becomes the vault **"Client 1 - JAG"**. A top-level + folder with no parent just uses its own name. If two different folders + would otherwise produce the same title, the script automatically extends + the title with more of the path until it's unique. +3. **Classifies each entry** into the closest matching 1Password item + category (Login, Secure Note, Credit Card, Identity, SSH Key, Password, + Document, API Credentials, Bank Account, Database, Driver License, Email, + Medical Record, Membership, Outdoor License, Passport, Rewards, Router, + Server, Social Security Number, Software License) based on the entry's + field names. See [How classification works](#how-classification-works). +4. **Creates the vaults and items** in 1Password (or reuses a vault that + already has the same title). Vaults are created up front; items are + imported in batches of up to 50 per vault via the SDK `create_all` API + (Document, SSH Key, and Credit Card items are created individually). +5. **Writes an import log** after a real import so you can review what was + created or roll it back later. + +## Setup + +```bash +pip install -r requirements.txt +``` + +Or install the dependency directly: + +```bash +pip install onepassword-sdk +``` + +Authenticate with **either** a service account **or** your signed-in 1Password +desktop app. + +### Option A: Service account + +Create a 1Password **Service Account** with permission to create vaults and +items: +https://developer.1password.com/docs/service-accounts/get-started + +```bash +export OP_SERVICE_ACCOUNT_TOKEN="ops_..." +``` + +### Option B: Desktop app (no service account) + +1. Install the [1Password desktop app](https://1password.com/downloads/). +2. Enable **Settings → Developer → Integrate with other apps**. +3. Set the account name shown in the app sidebar: + +```bash +export OP_ACCOUNT_NAME="My Team" +``` + +Or pass it per run with `--account "My Team"`. + +Do not set both `OP_SERVICE_ACCOUNT_TOKEN` and `OP_ACCOUNT_NAME` at the same +time. + +## Running on Windows + +The script runs on Windows the same way as on macOS or Linux. You need +**Python 3.10+** and the [1Password desktop app](https://1password.com/downloads/) +(if using desktop auth). + +1. Install Python from [python.org](https://www.python.org/downloads/windows/) + (check **Add python.exe to PATH** during setup), or use the **`py`** + launcher that ships with the Python installer. +2. Open **PowerShell** or **Command Prompt** and go to this folder: + +```powershell +cd C:\path\to\PPS +pip install -r requirements.txt +``` + +If `python` is not recognized, use `py -3` instead (e.g. `py -3 keepass_to_1password.py ...`). + +### Environment variables on Windows + +**PowerShell (current session):** + +```powershell +$env:OP_SERVICE_ACCOUNT_TOKEN = "ops_..." +# or, for desktop app auth: +$env:OP_ACCOUNT_NAME = "My Team" +``` + +**Command Prompt (current session):** + +```cmd +set OP_SERVICE_ACCOUNT_TOKEN=ops_... +set OP_ACCOUNT_NAME=My Team +``` + +You can also pass desktop account name per run with `--account "My Team"` and +skip setting `OP_ACCOUNT_NAME`. + +For desktop auth, install and unlock the 1Password desktop app, then enable +**Settings → Developer → Integrate with other apps** before running the +import. + +### Example commands (Windows) + +```powershell +# Preview the plan (no auth required) +py -3 keepass_to_1password.py Export.xml --list-only + +# Dry run +py -3 keepass_to_1password.py Export.xml --dry-run --account "My Team" + +# Import (writes .\Export.pps-import.json in the current directory) +py -3 keepass_to_1password.py C:\Exports\Export.xml --account "My Team" + +# Roll back a previous import +py -3 keepass_to_1password.py --delete-import .\Export.pps-import.json --account "My Team" +``` + +Use forward slashes or quoted paths if filenames contain spaces. The import +log is always written to whatever directory you run the command from (not +next to the XML file). + +## Usage + +Examples below use `python`; on Windows, use `py -3` if `python` is not on +your PATH (see [Running on Windows](#running-on-windows)). + +```bash +# See what would happen — no SDK, no network, no auth required +python keepass_to_1password.py Export.xml --list-only + +# Preview the import, including vault existence checks via the SDK +python keepass_to_1password.py Export.xml --dry-run + +# Import (writes ./Export.pps-import.json in the current directory by default) +python keepass_to_1password.py Export.xml + +# Import with desktop app auth +python keepass_to_1password.py Export.xml --account "My Team" + +# Custom import log path +python keepass_to_1password.py Export.xml --log-file /path/to/import-log.json +``` + +Try it first against the bundled synthetic file, `Sample_Export.xml`, which +exercises every supported item category with fictional data: + +```bash +python keepass_to_1password.py Sample_Export.xml --list-only +python keepass_to_1password.py Sample_Export.xml --dry-run +``` + +### Rolling back an import + +After a real import, the script writes a JSON log in the **current working +directory** (default: `./.pps-import.json`) listing every +vault and item it created. Use `--delete-import` to remove that data: + +```bash +# Preview what would be deleted +python keepass_to_1password.py Export.xml --delete-import --dry-run + +# Delete vaults/items recorded in ./Export.pps-import.json +python keepass_to_1password.py Export.xml --delete-import + +# Or point directly at the log file +python keepass_to_1password.py --delete-import Export.pps-import.json +``` + +Cleanup behavior: + +- **Vaults the script created** are deleted entirely (including all items in + them). +- **Vaults that already existed** are left in place; only the individual + items added by the import are deleted. +- If cleanup completes successfully, the log file is removed automatically. + +The log file contains vault and item IDs — treat it like sensitive data and +do not commit it to source control. + +## How classification works + +Pleasant Password Server exports have no built-in concept of "item type" — +every entry is just a title/username/password/URL/notes plus whatever custom +fields you or a tool added. There's no universal standard for naming those +custom fields, so this script uses a **best-effort heuristic**: it looks at +the *names* of an entry's custom fields (case- and spacing-insensitive) and +matches them against signatures for each 1Password category. For example, an +entry with `Card Number` and `CVV` fields is classified as a Credit Card; an +entry with only `Notes` and nothing else becomes a Secure Note; an entry +with `SSH Private Key` becomes an SSH Key item. + +**This is not guaranteed to be perfect.** If your export entries use +different field-naming conventions than the ones listed below, entries will +fall back to Login (if they have a username/password), Password (password +only, no username), or Secure Note (notes only). You can always review and +recategorize items afterward in 1Password — nothing is deleted from the +source, and no field data is discarded (see [What happens to +data that doesn't fit a known field](#what-happens-to-data-that-doesnt-fit-a-known-field) +below). + +### Recognized field-name signatures + +| 1Password category | Trigger field names (normalized) | +|---|---| +| SSH Key | `SSH Private Key` | +| Crypto Wallet | `Wallet Address`, `Recovery Phrase`, `Seed Phrase`, `Private Key (WIF)`, `Cryptocurrency` | +| Credit Card | `Card Number`, `CVV`, `Cardholder Name`, `Card Type` | +| Bank Account | `Account Number`, `Routing Number`, `IBAN`, `Bank Name` | +| Social Security Number | `SSN`, `Social Security Number` | +| Passport | `Passport Number`, `Nationality` | +| Driver License | `License Number`, `License State`, `License Class` | +| Software License | `License Key`, `Serial Number`, `Licensed To` | +| Outdoor License | `Permit Number`, `Hunting Season`, `Game Zone` | +| Medical Record | `Blood Type`, `Policy Number`, `Physician Name`, `Medical Conditions` | +| Membership | `Membership Number`, `Membership Level` | +| Rewards | `Rewards Number`, `Points Balance`, `Tier Status` | +| API Credentials | `API Key`, `Client ID`, `Client Secret`, `Access Token` | +| Database | any two of `Database Name`, `Hostname`, `Port` | +| Server | `IP Address`, `OS Version`, `Server Name` | +| Router | `SSID`, `WiFi Password`, `Router Admin URL` | +| Identity | any two of `First Name`, `Last Name`, `Date of Birth`, `Address`, `City` | +| Email | `IMAP Server`, `POP3 Server`, `SMTP Server`, `Email Address` | +| Secure Note (contact) | `Full Name` or `Relationship` — see note below | +| Document | entry has a file attachment in the export | +| Secure Note | only `Notes` is populated (no username/password/URL) | +| Password | `Password` populated, no `UserName` | +| Login | default fallback | + +**Contact entries:** Export entries with `Full Name` / `Relationship` fields +are classified as contact-style data, but the SDK cannot create native +**Person** items. They are imported as **Secure Notes** with the contact +fields preserved in a Details section. + +A one-time password (`otp`/`TOTP` field, including an `otpauth://` URI) is +detected on any entry and added as a proper TOTP field regardless of +category. + +### Import fallbacks + +If a specific item type cannot be created, the script retries as a Secure +Note rather than stopping the whole import: + +- **SSH Key** — unparseable or placeholder key material +- **Document** — attachment could not be attached as a native Document item +- **Other structured categories** — SDK rejects the item payload + +Failed items are also recorded in the import log under `failures`. + +### What happens to data that doesn't fit a known field + +- Recognized fields (the table above) are mapped to an appropriately-typed + 1Password field (e.g. card number → Credit Card Number field, expiry date + → Month/Year field, SSN/CVV/API keys → Concealed field) inside a + **"Details"** section. +- Any other custom field from the export is preserved as a Text or Concealed + field (guessed from the field name — anything containing "password", + "secret", "key", "pin", "cvv", "ssn", "private", or "token" is treated as + a secret) inside an **"Export metadata"** section, so nothing is silently + dropped. +- `Notes` always carries over as the item's Notes field. + +## Flags + +| Flag | Effect | +|---|---| +| `--list-only` | Parse and print the vault/item/category plan. No SDK, no network, no auth required. | +| `--dry-run` | Preview actions via the SDK (vault checks on import; deletion preview on cleanup). Creates nothing. Requires the SDK and valid auth. | +| `--account NAME` | 1Password account name for desktop app auth (overrides `OP_ACCOUNT_NAME`). | +| `--log-file PATH` | Import log path (default: `./.pps-import.json` in the current directory). | +| `--delete-import` | Delete vaults/items from a previous import log instead of importing. Use with `--dry-run` to preview. | +| (no flag) | Run the import and write the log file. | + +`--list-only` and `--dry-run` cannot be combined. `--delete-import` cannot +be used with `--list-only`. + +## Import log format + +The log is JSON with this structure: + +```json +{ + "source_xml": "/path/to/Export.xml", + "imported_at": "2026-01-15T12:00:00+00:00", + "vaults": [ + { + "title": "Client 1 - JAG", + "id": "vault-id", + "created": true, + "items": [ + { + "title": "Example Login", + "id": "item-id", + "category": "Login", + "planned_category": "Login" + } + ] + } + ], + "failures": [] +} +``` + +- `created: true` — the script created this vault; `--delete-import` removes + the whole vault. +- `created: false` — the script reused an existing vault; `--delete-import` + removes only the listed items. + +## Notes and limitations + +- **Vault titles must be unique in a 1Password account.** If a vault with + the computed title already exists, the script reuses it instead of + creating a duplicate. +- **Attachments:** only the first file attached to an export entry is + imported, as the item's single Document file, and only for entries + classified as Document. Attachments on entries of other categories aren't + currently uploaded as field-level files. +- **Address fields** are stored as plain text (street/city/etc. as separate + text fields) rather than 1Password's structured Address field type, to + keep the mapping simple and predictable. +- Export `TOTPDigits`/`TOTPPeriod`/etc. settings (as opposed to an actual + seed) carry over into the Export metadata section if present, since + 1Password derives digit/period info from the TOTP seed or `otpauth://` URI + itself. +- **Re-running an import** against the same export will reuse existing vaults + by title and may create duplicate items. Use the import log and + `--delete-import` to clean up test runs. +- **Batch import:** most items are created with `items.create_all()` in chunks + of 50. Document, SSH Key, and Credit Card items are always created one at + a time because they carry binary data or need special SDK handling. If a + batch call fails for a single item, the script retries that item + individually (including Secure Note fallbacks). + +## Files in this folder + +- `keepass_to_1password.py` — the importer (parsing, classification, + 1Password import, logging, and cleanup). +- `requirements.txt` — Python dependencies (`onepassword-sdk`). +- `Sample_Export.xml` — a synthetic Pleasant Password Server export with + entirely fictional data (fake names, `example.com` addresses, RFC 5737 test + IP ranges, the well-known `4111111111111111` test Visa number, etc.) + covering every supported item category, for trying the script out safely. + +## Security note + +A real Pleasant Password Server export contains live plaintext passwords +once decrypted to XML. Treat the export file, import logs, and anything +derived from them as secrets: avoid committing them to source control, +delete them once the import is complete, and don't leave copies lying around +in shared folders. diff --git a/1password/migration/PPS/Sample_Export.xml b/1password/migration/PPS/Sample_Export.xml new file mode 100644 index 0000000..f44611a --- /dev/null +++ b/1password/migration/PPS/Sample_Export.xml @@ -0,0 +1,347 @@ + + + + Pleasant Password Server + Sample Export + Synthetic test data - no real people, companies, or credentials + + False + False + True + False + False + + + U2FtcGxlIGRvY3VtZW50IGF0dGFjaG1lbnQgZm9yIFBsZWFzYW50IFBhc3N3b3JkIFNlcnZlciB0byAxUGFzc3dvcmQgaW1wb3J0IHRlc3RpbmcuCg== + + + + + Fabrikam Vault Export + + IT Operations + + Network Team + + AAAAAAAAAAAAAAAAAAAAAA== + + 2026-01-05T10:00:00Z + 2026-01-05T10:00:00Z + + TitleBranch Office Wi-Fi Router + UserNameadmin + PasswordRtr-Sample-Pw-01 + URLhttps://192.0.2.1/admin + NotesTest router entry for classifier validation. + SSIDFABRIKAM-GUEST + WiFiPasswordGuestNetSample123 + RouterAdminURLhttps://192.0.2.1/admin + + + AAAAAAAAAAAAAAAAAAAAAQ== + + 2026-01-05T10:05:00Z + 2026-01-05T10:05:00Z + + TitleApp Server 01 + UserNamesvc-appserver + PasswordSrv-Sample-Pw-02 + NotesTest server entry for classifier validation. + IPAddress198.51.100.10 + OSVersionUbuntu 24.04 LTS + + + AAAAAAAAAAAAAAAAAAAAAg== + + 2026-01-05T10:10:00Z + 2026-01-05T10:10:00Z + + TitleReporting Database + UserNamedb_reporting + PasswordDb-Sample-Pw-03 + NotesTest database entry for classifier validation. + Hostnamereporting-db.example.internal + DatabaseNamereporting + Port5432 + + + AAAAAAAAAAAAAAAAAAAAAw== + + 2026-01-05T10:15:00Z + 2026-01-05T10:15:00Z + + TitleDeploy Bot SSH Key + UserNamedeploybot + NotesTest SSH key entry for classifier validation. Key material below is a placeholder, not a real key. + SSHPrivateKey-----BEGIN OPENSSH PRIVATE KEY----- +b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAAAMwAAAAtzc2gtZW +QyNTUxOQAAACBmYLT61Pai4KZmiXUXKSnEjcyNTq5cVcGA6v1zi0SQggAAAKB42E6WeNhO +lgAAAAtzc2gtZWQyNTUxOQAAACBmYLT61Pai4KZmiXUXKSnEjcyNTq5cVcGA6v1zi0SQgg +AAAEBNEOQCfFCwXJK1ijJK9PPosAOD+GcRtMFESkoN6hKP9mZgtPrU9qLgpmaJdRcpKcSN +zI1OrlxVwYDq/XOLRJCCAAAAFnNhbXBsZS1vbmx5LWRvLW5vdC11c2UBAgMEBQYH +-----END OPENSSH PRIVATE KEY----- + KeyTypeed25519 + + + AAAAAAAAAAAAAAAAAAAABA== + + 2026-01-05T10:20:00Z + 2026-01-05T10:20:00Z + + TitleIntranet Portal + UserNameportal.user + PasswordPortal-Sample-Pw-04 + URLhttps://intranet.example.com + NotesPlain generic login entry, no special fields. + + + + Security Team + + AAAAAAAAAAAAAAAAAAAABQ== + + 2026-01-06T09:00:00Z + 2026-01-06T09:00:00Z + + TitleBilling Service API Credentials + NotesTest API credentials entry for classifier validation. + APIKeysample-api-key-0000000000000000 + ClientIDsample-client-id-1234 + ClientSecretsample-client-secret-abcd + + + AAAAAAAAAAAAAAAAAAAABg== + + 2026-01-06T09:05:00Z + 2026-01-06T09:05:00Z + + TitleServer Room Door Code + Password024681 + NotesPassword-only entry, no username, for classifier validation. + + + AAAAAAAAAAAAAAAAAAAABw== + + 2026-01-06T09:10:00Z + 2026-01-06T09:10:00Z + + TitleIncident Response Runbook + Notes1. Identify scope. +2. Notify on-call. +3. Contain and eradicate. +4. Document timeline. +This is a secure-note-only test entry with no username, password, or URL. + + + AAAAAAAAAAAAAAAAAAAACA== + + 2026-01-06T09:15:00Z + 2026-01-06T09:15:00Z + + TitleAdmin Console Login + UserNameconsole.admin + PasswordConsole-Sample-Pw-05 + URLhttps://console.example.com + NotesLogin entry with a TOTP seed for classifier validation. + otpotpauth://totp/Sample:console.admin?secret=JBSWY3DPEHPK3PXP&issuer=Sample + + + AAAAAAAAAAAAAAAAAAAACQ== + + 2026-01-06T09:20:00Z + 2026-01-06T09:20:00Z + + TitleOnboarding Checklist + NotesDocument item test entry - see attached file. + + onboarding-checklist.txt + + + + + + + Personal + + Family Records + + AAAAAAAAAAAAAAAAAAAACg== + + 2026-01-07T08:00:00Z + 2026-01-07T08:00:00Z + + TitleSample Person Identity + NotesIdentity test entry for classifier validation. + FirstNameAlex + LastNameSampleton + DateOfBirth1990-01-01 + Address123 Test Street + CitySampleville + EmailAddressalex.sampleton@example.com + + + AAAAAAAAAAAAAAAAAAAACw== + + 2026-01-07T08:05:00Z + 2026-01-07T08:05:00Z + + TitleSample Passport + NotesPassport test entry for classifier validation. + PassportNumberX0000000 + NationalityTestland + DateOfIssue2020-01-01 + DateOfExpiry2030-01-01 + + + AAAAAAAAAAAAAAAAAAAADA== + + 2026-01-07T08:10:00Z + 2026-01-07T08:10:00Z + + TitleSample Driver License + NotesDriver license test entry for classifier validation. + LicenseNumberD0000-00000-00000 + LicenseStateTest State + LicenseClassC + + + AAAAAAAAAAAAAAAAAAAADQ== + + 2026-01-07T08:15:00Z + 2026-01-07T08:15:00Z + + TitleSample SSN Record + NotesSSN test entry for classifier validation. + SocialSecurityNumber000-12-3456 + + + AAAAAAAAAAAAAAAAAAAADg== + + 2026-01-07T08:20:00Z + 2026-01-07T08:20:00Z + + TitleSample Medical Record + NotesMedical record test entry for classifier validation. + PolicyNumberPOL-SAMPLE-0001 + BloodTypeO+ + PhysicianNameDr. Sample Physician + + + AAAAAAAAAAAAAAAAAAAADw== + + 2026-01-07T08:25:00Z + 2026-01-07T08:25:00Z + + TitleSample Emergency Contact + NotesPerson/contact test entry for classifier validation. + FullNameJordan Sampleton + RelationshipSibling + PhoneNumber+1-555-0100 + + + + Shopping and Memberships + + AAAAAAAAAAAAAAAAAAAAEA== + + 2026-01-08T07:00:00Z + 2026-01-08T07:00:00Z + + TitleSample Visa Card + NotesCredit card test entry for classifier validation. Uses a well-known test card number. + CardholderNameAlex Sampleton + CardNumber4111111111111111 + ExpirationDate01/2030 + CVV123 + CardTypeVisa + + + AAAAAAAAAAAAAAAAAAAAEQ== + + 2026-01-08T07:05:00Z + 2026-01-08T07:05:00Z + + TitleSample Checking Account + NotesBank account test entry for classifier validation. + BankNameSample Test Bank + AccountNumber000123456789 + RoutingNumber011000015 + + + AAAAAAAAAAAAAAAAAAAAEg== + + 2026-01-08T07:10:00Z + 2026-01-08T07:10:00Z + + TitleSample Crypto Wallet + NotesCrypto wallet test entry for classifier validation. + WalletAddress0x0000000000000000000000000000000000dEaD + RecoveryPhrasesample test seed phrase words go here only for testing purposes + CryptocurrencyETH (testnet) + + + AAAAAAAAAAAAAAAAAAAAEw== + + 2026-01-08T07:15:00Z + 2026-01-08T07:15:00Z + + TitleSample Email Account + UserNamesample.user@example.com + PasswordEmail-Sample-Pw-06 + NotesEmail account test entry for classifier validation. + IMAPServerimap.example.com + SMTPServersmtp.example.com + + + AAAAAAAAAAAAAAAAAAAAFA== + + 2026-01-08T07:20:00Z + 2026-01-08T07:20:00Z + + TitleSample Gym Membership + NotesMembership test entry for classifier validation. + MembershipNumberMEM-SAMPLE-0002 + MembershipLevelGold + + + AAAAAAAAAAAAAAAAAAAAFQ== + + 2026-01-08T07:25:00Z + 2026-01-08T07:25:00Z + + TitleSample Airline Rewards + NotesRewards test entry for classifier validation. + RewardsNumberRWD-SAMPLE-0003 + PointsBalance12345 + TierStatusSilver + + + AAAAAAAAAAAAAAAAAAAAFg== + + 2026-01-08T07:30:00Z + 2026-01-08T07:30:00Z + + TitleSample Design Software License + NotesSoftware license test entry for classifier validation. + LicenseKeySAMPLE-0000-0000-0000 + SerialNumberSN-SAMPLE-0004 + LicensedToAlex Sampleton + + + AAAAAAAAAAAAAAAAAAAAFw== + + 2026-01-08T07:35:00Z + 2026-01-08T07:35:00Z + + TitleSample Fishing Permit + NotesOutdoor license test entry for classifier validation. + PermitNumberPERMIT-SAMPLE-0005 + HuntingSeason2026 + GameZoneZone 7 + + + + + + + diff --git a/1password/migration/PPS/keepass_to_1password.py b/1password/migration/PPS/keepass_to_1password.py new file mode 100644 index 0000000..dc0b29e --- /dev/null +++ b/1password/migration/PPS/keepass_to_1password.py @@ -0,0 +1,1052 @@ +#!/usr/bin/env python3 +""" +keepass_to_1password.py +======================== + +Parse a Pleasant Password Server XML export and import it straight into +1Password using the official 1Password Python SDK (`onepassword-sdk` on PyPI), +in one step. + +- Every Pleasant Password Server Group that directly contains entries becomes + one 1Password vault. The vault is named " - " + (e.g. a path of .../JAG/Client 1 becomes the vault "Client 1 - JAG"). + Top-level groups with no parent just use their own name. +- Every export Entry is mapped to the closest matching 1Password item + category (Login, Secure Note, Credit Card, Identity, SSH Key, Password, + Document, etc.) using the entry's field names as signals. See + README.md for how the classifier works and its limitations. + +Setup +----- + pip install -r requirements.txt + +Authenticate with either a service account token or your signed-in +1Password desktop app (Settings → Developer → Integrate with other apps): + + export OP_SERVICE_ACCOUNT_TOKEN="ops_..." # service account + # or + export OP_ACCOUNT_NAME="My Team" # desktop app account name (sidebar label) + +Usage +----- + python keepass_to_1password.py Export.xml + python keepass_to_1password.py Export.xml --dry-run + python keepass_to_1password.py Export.xml --account "My Team" + python keepass_to_1password.py Sample_Export.xml --dry-run # try it on the bundled sample first + python keepass_to_1password.py Export.xml --log-file Export.pps-import.json + python keepass_to_1password.py --delete-import Export.pps-import.json + python keepass_to_1password.py Export.xml --delete-import # uses ./Export.pps-import.json +""" + +from __future__ import annotations + +import argparse +import asyncio +import base64 +import gzip +import json +import os +import re +import sys +from datetime import datetime, timezone +from pathlib import Path +import xml.etree.ElementTree as ET +from collections import defaultdict +from dataclasses import dataclass +from typing import Any, Dict, List, Optional + + +# -------------------------------------------------------------------------- +# Pleasant Password Server XML parsing (KeePass XML 2.x format) +# -------------------------------------------------------------------------- + +def normalize_key(key: str) -> str: + """'Card Number' -> 'cardnumber' so we can match field names loosely.""" + return re.sub(r"[^a-z0-9]", "", key.lower()) + + +def text_of(el, tag, default=""): + child = el.find(tag) + if child is None or child.text is None: + return default + return child.text + + +def parse_binaries(root) -> Dict[str, bytes]: + """Return {binary_id: raw bytes} from Meta/Binaries, decompressing gzip + payloads when Compressed="True", as KeePass XML 2.x exports do.""" + binaries = {} + for bin_el in root.findall("./Meta/Binaries/Binary"): + bin_id = bin_el.get("ID") + if bin_id is None or bin_el.text is None: + continue + raw = base64.b64decode(bin_el.text) + if bin_el.get("Compressed", "False").lower() == "true": + try: + raw = gzip.decompress(raw) + except OSError: + pass # leave as-is if it isn't actually gzip + binaries[bin_id] = raw + return binaries + + +def parse_entry(entry_el, binaries: Dict[str, bytes]) -> dict: + fields = {} + for string_el in entry_el.findall("String"): + key_el = string_el.find("Key") + val_el = string_el.find("Value") + if key_el is None: + continue + key = key_el.text or "" + value = val_el.text if val_el is not None and val_el.text is not None else "" + fields[key] = value + + attachments = [] + for bin_el in entry_el.findall("Binary"): + fname = text_of(bin_el, "Key", "attachment") + value_el = bin_el.find("Value") + ref = value_el.get("Ref") if value_el is not None else None + if ref is not None and ref in binaries: + attachments.append({"name": fname, "content": binaries[ref]}) + + return { + "raw_fields": fields, # original export String key -> value + "attachments": attachments, # [{name, content bytes}] + } + + +def walk_group(group_el, path: List[str], binaries, vaults: dict): + name = text_of(group_el, "Name", "(unnamed)") + current_path = path + [name] + + entries = [parse_entry(e, binaries) for e in group_el.findall("Entry")] + if entries: + key = " / ".join(current_path) + vaults[key]["path"] = current_path + vaults[key]["entries"].extend(entries) + + for sub in group_el.findall("Group"): + walk_group(sub, current_path, binaries, vaults) + + +def vault_title_for(path: List[str]) -> str: + """'Last path value - second last value', e.g. .../JAG/Client 1 -> 'Client 1 - JAG'.""" + if len(path) >= 2: + return f"{path[-1]} - {path[-2]}" + return path[-1] + + +def disambiguate_titles(vaults: dict): + """If two different folders would produce the same 'leaf - parent' title, + extend with more of the path (grandparent, etc.) until unique.""" + def title_at_depth(path, depth): + segment = path[-depth:] if depth <= len(path) else path[:] + if len(segment) >= 2: + return f"{segment[-1]} - {' / '.join(reversed(segment[:-1]))}" + return segment[-1] + + depth = 2 + while True: + seen = defaultdict(list) + for v in vaults.values(): + seen[title_at_depth(v["path"], depth)].append(v) + if all(len(vs) == 1 for vs in seen.values()): + for title, vs in seen.items(): + vs[0]["vault_title"] = title + return + max_len = max(len(v["path"]) for v in vaults.values()) + if depth >= max_len: + # last resort: fully qualify, plus a numeric suffix for exact dupes + counts = defaultdict(int) + for v in vaults.values(): + full = " / ".join(v["path"]) + counts[full] += 1 + v["vault_title"] = full if counts[full] == 1 else f"{full} ({counts[full]})" + return + depth += 1 + + +def parse_pps_xml(xml_path: str) -> List[dict]: + tree = ET.parse(xml_path) + root = tree.getroot() + root_group = root.find("Root/Group") + if root_group is None: + print("Could not find Root/Group in the XML file.", file=sys.stderr) + sys.exit(1) + + binaries = parse_binaries(root) + vaults = defaultdict(lambda: {"path": [], "entries": []}) + walk_group(root_group, [], binaries, vaults) + disambiguate_titles(vaults) + return list(vaults.values()) + + +# -------------------------------------------------------------------------- +# Export entry -> 1Password category classifier +# -------------------------------------------------------------------------- +# Pleasant Password Server exports have no native concept of "item type" - +# every entry is just a bag of Title/UserName/Password/URL/Notes plus arbitrary +# custom String fields. To recover something like 1Password's item categories, +# we look for field-name signatures commonly attached to export entries +# (e.g. "Card Number" + "CVV" strongly implies a credit card). This is a +# best-effort heuristic, not a guarantee - see README.md. + +# Each rule: (category name, {normalized field-name signals}, min matches needed) +CATEGORY_RULES = [ + ("Document", set(), 0), # handled separately (has attachment) + ("SshKey", {"sshprivatekey"}, 1), + ("CryptoWallet", {"walletaddress", "recoveryphrase", "seedphrase", "privatekeywif", "cryptocurrency"}, 1), + ("CreditCard", {"cardnumber", "cvv", "cardholdername", "cardtype"}, 1), + ("BankAccount", {"accountnumber", "routingnumber", "iban", "bankname"}, 1), + ("SocialSecurityNumber", {"ssn", "socialsecuritynumber"}, 1), + ("Passport", {"passportnumber", "nationality"}, 1), + ("DriverLicense", {"licensenumber", "licensestate", "licenseclass"}, 1), + ("SoftwareLicense", {"licensekey", "serialnumber", "licensedto"}, 1), + ("OutdoorLicense", {"permitnumber", "huntingseason", "gamezone"}, 1), + ("MedicalRecord", {"bloodtype", "policynumber", "physicianname", "medicalconditions"}, 1), + ("Membership", {"membershipnumber", "membershiplevel"}, 1), + ("Rewards", {"rewardsnumber", "pointsbalance", "tierstatus"}, 1), + ("ApiCredentials", {"apikey", "clientid", "clientsecret", "accesstoken"}, 1), + ("Database", {"databasename", "hostname", "port"}, 2), + ("Server", {"ipaddress", "osversion", "servername"}, 1), + ("Router", {"ssid", "wifipassword", "routeradminurl"}, 1), + ("Identity", {"firstname", "lastname", "dateofbirth", "address", "city"}, 2), + ("Email", {"imapserver", "pop3server", "smtpserver", "emailaddress"}, 1), + ("Person", {"fullname", "relationship"}, 1), +] + +DOCUMENT_KEY_HINTS = {"filename", "attachment", "document"} + + +def classify_entry(fields: dict, attachments: list) -> str: + if attachments: + return "Document" + + normalized = {normalize_key(k) for k in fields.keys() if fields.get(k)} + + for category, signals, min_matches in CATEGORY_RULES: + if category == "Document": + continue + if signals and len(normalized & signals) >= min_matches: + # The SDK rejects Person items outright ("bad request body"); preserve + # contact fields on a Secure Note instead. + if category == "Person": + return "SecureNote" + return category + + has_username = bool(fields.get("UserName")) + has_password = bool(fields.get("Password")) + has_url = bool(fields.get("URL")) + has_notes = bool(fields.get("Notes")) + + if not has_username and not has_password and not has_url and has_notes: + return "SecureNote" + if has_password and not has_username: + return "Password" + return "Login" # default fallback + + +# -------------------------------------------------------------------------- +# 1Password field construction +# -------------------------------------------------------------------------- + +STANDARD_KEYS = {"Title", "UserName", "Password", "URL", "Notes"} +TOTP_KEYS = {"otp", "totp", "totpseed", "totpseedbase32"} + +# normalized export field key -> (field id, display title, ItemFieldType member name) +# ItemFieldType member names are resolved against the SDK's actual enum at +# runtime so this table doubles as documentation of the mapping. +FIELD_TYPE_HINTS = { + "cardnumber": ("card_number", "card number", "CREDITCARDNUMBER"), + "cardtype": ("card_type", "card type", "CREDITCARDTYPE"), + "cvv": ("cvv", "verification number", "CONCEALED"), + "expirationdate": ("expiry", "expiry date", "MONTHYEAR"), + "cardholdername": ("cardholder", "cardholder name", "TEXT"), + "walletaddress": ("wallet_address", "wallet address", "TEXT"), + "recoveryphrase": ("recovery_phrase", "recovery phrase", "CONCEALED"), + "seedphrase": ("seed_phrase", "seed phrase", "CONCEALED"), + "privatekeywif": ("private_key", "private key", "CONCEALED"), + "cryptocurrency": ("currency", "currency", "TEXT"), + "apikey": ("api_key", "credential", "CONCEALED"), + "clientid": ("client_id", "client id", "TEXT"), + "clientsecret": ("client_secret", "client secret", "CONCEALED"), + "accesstoken": ("access_token", "access token", "CONCEALED"), + "accountnumber": ("account_number", "account number", "CONCEALED"), + "routingnumber": ("routing_number", "routing number", "TEXT"), + "iban": ("iban", "IBAN", "TEXT"), + "bankname": ("bank_name", "bank name", "TEXT"), + "hostname": ("hostname", "hostname", "TEXT"), + "databasename": ("database_name", "database", "TEXT"), + "port": ("port", "port", "TEXT"), + "ipaddress": ("ip_address", "IP address", "TEXT"), + "osversion": ("os_version", "operating system", "TEXT"), + "servername": ("server_name", "server name", "TEXT"), + "ssid": ("ssid", "network name (SSID)", "TEXT"), + "wifipassword": ("wifi_password", "Wi-Fi password", "CONCEALED"), + "routeradminurl": ("admin_url", "admin URL", "URL"), + "licensenumber": ("license_number", "license number", "CONCEALED"), + "licensestate": ("license_state", "issuing state/region", "TEXT"), + "licenseclass": ("license_class", "license class", "TEXT"), + "passportnumber": ("passport_number", "passport number", "CONCEALED"), + "nationality": ("nationality", "nationality", "TEXT"), + "dateofissue": ("date_of_issue", "date of issue", "TEXT"), + "dateofexpiry": ("date_of_expiry", "date of expiry", "TEXT"), + "ssn": ("ssn", "SSN", "CONCEALED"), + "socialsecuritynumber": ("ssn", "SSN", "CONCEALED"), + "licensekey": ("license_key", "license key", "CONCEALED"), + "serialnumber": ("serial_number", "serial number", "TEXT"), + "licensedto": ("licensed_to", "licensed to", "TEXT"), + "permitnumber": ("permit_number", "permit number", "TEXT"), + "huntingseason": ("season", "season", "TEXT"), + "gamezone": ("zone", "zone", "TEXT"), + "bloodtype": ("blood_type", "blood type", "TEXT"), + "policynumber": ("policy_number", "policy number", "TEXT"), + "physicianname": ("physician", "physician", "TEXT"), + "medicalconditions": ("conditions", "conditions", "TEXT"), + "membershipnumber": ("membership_number", "membership number", "TEXT"), + "membershiplevel": ("membership_level", "membership level", "TEXT"), + "rewardsnumber": ("rewards_number", "rewards number", "TEXT"), + "pointsbalance": ("points_balance", "points balance", "TEXT"), + "tierstatus": ("tier_status", "tier status", "TEXT"), + "firstname": ("first_name", "first name", "TEXT"), + "lastname": ("last_name", "last name", "TEXT"), + "dateofbirth": ("date_of_birth", "date of birth", "TEXT"), + "address": ("address", "address", "TEXT"), + "city": ("city", "city", "TEXT"), + "emailaddress": ("email_address", "email address", "EMAIL"), + "fullname": ("full_name", "full name", "TEXT"), + "relationship": ("relationship", "relationship", "TEXT"), + "phonenumber": ("phone_number", "phone number", "PHONE"), + "imapserver": ("imap_server", "IMAP server", "TEXT"), + "pop3server": ("pop3_server", "POP3 server", "TEXT"), + "smtpserver": ("smtp_server", "SMTP server", "TEXT"), + "keytype": ("key_type", "key type", "TEXT"), +} + +SECRET_HINT_WORDS = ("password", "secret", "key", "pin", "cvv", "ssn", "private", "token") + + +def build_item(sdk, category_name: str, entry: dict, vault_id: str): + """Build an ItemCreateParams for one parsed export entry.""" + fields_raw = entry["raw_fields"] + category = getattr(sdk.ItemCategory, category_name.upper()) + + item_fields = [] + sections = [] + seen_normalized = set() + + title = fields_raw.get("Title", "").strip() or "(untitled)" + notes = fields_raw.get("Notes") or None + username = fields_raw.get("UserName") or "" + password = fields_raw.get("Password") or "" + url = fields_raw.get("URL") or "" + + # Login-style built-in fields for categories where a username/password + # pairing makes sense. + if category_name in ("Login", "Password") : + if username: + item_fields.append(sdk.ItemField(id="username", title="username", + field_type=sdk.ItemFieldType.TEXT, value=username)) + if password: + item_fields.append(sdk.ItemField(id="password", title="password", + field_type=sdk.ItemFieldType.CONCEALED, value=password)) + elif username or password: + # Other categories: still preserve username/password as plain fields + # rather than dropping them, since export entries can mix a login + # with category-specific fields. + if username: + item_fields.append(sdk.ItemField(id="username", title="username", + field_type=sdk.ItemFieldType.TEXT, value=username)) + if password: + item_fields.append(sdk.ItemField(id="password", title="password", + field_type=sdk.ItemFieldType.CONCEALED, value=password)) + + seen_normalized.update({"title", "notes", "username", "password", "url"}) + + # TOTP + totp_value = "" + for key, value in fields_raw.items(): + if normalize_key(key) in TOTP_KEYS and value: + totp_value = value + break + if totp_value: + sections.append(sdk.ItemSection(id="totpsection", title="One-Time Password")) + item_fields.append(sdk.ItemField( + id="onetimepassword", title="one-time password", + field_type=sdk.ItemFieldType.TOTP, section_id="totpsection", value=totp_value, + )) + seen_normalized.update({normalize_key(k) for k in TOTP_KEYS}) + + # SSH private key -> dedicated SSHKEY field (SshKey category only) + if category_name == "SshKey": + for key, value in fields_raw.items(): + if normalize_key(key) == "sshprivatekey" and value: + item_fields.append(sdk.ItemField( + id="private_key", title="private key", + field_type=sdk.ItemFieldType.SSHKEY, value=value, + )) + seen_normalized.add("sshprivatekey") + break + + # Category-specific known fields, in a dedicated section + known_section_added = False + for key, value in fields_raw.items(): + if not value: + continue + norm = normalize_key(key) + if norm in seen_normalized or key in STANDARD_KEYS: + continue + hint = FIELD_TYPE_HINTS.get(norm) + if hint: + field_id, field_title, type_name = hint + if not known_section_added: + sections.append(sdk.ItemSection(id="details", title="Details")) + known_section_added = True + field_type = getattr(sdk.ItemFieldType, type_name, sdk.ItemFieldType.TEXT) + item_fields.append(sdk.ItemField( + id=field_id, title=field_title, field_type=field_type, + section_id="details", value=value, + )) + seen_normalized.add(norm) + + # Anything left over: dump into an "Export metadata" section so nothing + # is silently lost, guessing CONCEALED vs TEXT from the key name. + leftover_added = False + for i, (key, value) in enumerate(fields_raw.items()): + norm = normalize_key(key) + if key in STANDARD_KEYS or norm in seen_normalized or not value: + continue + if not leftover_added: + sections.append(sdk.ItemSection(id="pps_meta", title="Export metadata")) + leftover_added = True + is_secret = any(w in norm for w in SECRET_HINT_WORDS) + item_fields.append(sdk.ItemField( + id=f"kp_{i}", title=key, + field_type=sdk.ItemFieldType.CONCEALED if is_secret else sdk.ItemFieldType.TEXT, + section_id="pps_meta", value=str(value), + )) + + websites = None + if url and category_name in ("Login", "Password"): + websites = [sdk.Website(url=url, label="website", + autofill_behavior=sdk.AutofillBehavior.ANYWHEREONWEBSITE)] + elif url: + # keep the URL even for categories that don't support autofill websites + item_fields.append(sdk.ItemField(id="url", title="URL", + field_type=sdk.ItemFieldType.URL, value=url)) + + document = None + if category_name == "Document": + if entry["attachments"]: + att = entry["attachments"][0] # Document items hold a single primary file + document = sdk.DocumentCreateParams(name=att["name"], content=att["content"]) + if len(entry["attachments"]) > 1: + extra = ", ".join(a["name"] for a in entry["attachments"][1:]) + notes = (notes or "") + f"\n\n[Additional export attachments not imported: {extra}]" + else: + notes = (notes or "") + "\n\n[Classified as Document but no attachment was found in the export.]" + category = sdk.ItemCategory.SECURENOTE + category_name = "SecureNote" + + kwargs = dict( + title=title, + category=category, + vault_id=vault_id, + fields=item_fields or None, + sections=sections or None, + notes=notes, + websites=websites, + ) + if document: + kwargs["document"] = document + + return sdk.ItemCreateParams(**kwargs) + + +def _ssh_key_import_error(exc: Exception) -> bool: + msg = str(exc).lower() + return "private key" in msg or "ssh key" in msg or "openssh" in msg + + +def _document_import_error(exc: Exception) -> bool: + msg = str(exc).lower() + return "document" in msg and ("file" in msg or "attach" in msg) + + +def _structured_item_error(exc: Exception) -> bool: + msg = str(exc).lower() + return "bad request body" in msg or "bad input passed by the user" in msg + + +async def create_item(client, sdk, category_name: str, entry: dict, vault_id: str, item_title: str): + """Create an item, with Secure Note fallbacks for unparseable SSH keys or documents.""" + params = build_item(sdk, category_name, entry, vault_id) + try: + created = await client.items.create(params) + return created, category_name + except Exception as exc: + note = entry["raw_fields"].get("Notes") or "" + if category_name == "SshKey" and _ssh_key_import_error(exc): + print( + f" warning: could not parse SSH key for {item_title!r} ({exc}); " + "saving as Secure Note with concealed key text", + file=sys.stderr, + ) + fallback_note = ( + f"{note}\n\n[SSH key could not be imported as a native SSH Key item; " + "stored as concealed text instead.]" + ).strip() + elif category_name == "Document" and _document_import_error(exc): + att_names = ", ".join(a["name"] for a in entry["attachments"]) or "(unknown)" + print( + f" warning: could not attach document for {item_title!r} ({exc}); " + "saving as Secure Note", + file=sys.stderr, + ) + fallback_note = ( + f"{note}\n\n[Document attachment(s) could not be imported: {att_names}.]" + ).strip() + elif _structured_item_error(exc): + print( + f" warning: could not create {category_name} item {item_title!r} ({exc}); " + "saving as Secure Note", + file=sys.stderr, + ) + fallback_note = ( + f"{note}\n\n[{category_name} item could not be created via the SDK; " + "stored as a Secure Note instead.]" + ).strip() + else: + raise + + fallback_entry = { + **entry, + "raw_fields": {**entry["raw_fields"], "Notes": fallback_note}, + "attachments": [], + } + params = build_item(sdk, "SecureNote", fallback_entry, vault_id) + created = await client.items.create(params) + return created, "SecureNote" + + +BULK_CREATE_MAX = 50 + +# Item types that must be created individually (binary payloads or SDK constraints). +INDIVIDUAL_CREATE_CATEGORIES = frozenset({"Document", "SshKey", "CreditCard"}) + + +@dataclass +class _PendingItem: + entry: dict + category_name: str + item_title: str + params: Any + + +def _chunked(items: List, size: int) -> List[List]: + return [items[i: i + size] for i in range(0, len(items), size)] + + +def needs_individual_create(category_name: str, params: Any) -> bool: + if category_name in INDIVIDUAL_CREATE_CATEGORIES: + return True + return getattr(params, "document", None) is not None + + +def _record_created_item( + created, + actual_category: str, + planned_category: str, + vault_log: Optional[dict], + category_counts: dict, +) -> None: + if actual_category != planned_category: + category_counts[planned_category] -= 1 + category_counts[actual_category] += 1 + print(f" created {actual_category} item {created.title!r} (id={created.id})") + if vault_log is not None: + vault_log["items"].append({ + "title": created.title, + "id": created.id, + "category": actual_category, + "planned_category": planned_category, + }) + + +async def _create_one_item( + client, + sdk, + pending: _PendingItem, + vault_id: str, + vault_title: str, + vault_log: Optional[dict], + import_log: Optional[dict], + category_counts: dict, +) -> bool: + try: + created, actual_category = await create_item( + client, sdk, pending.category_name, pending.entry, vault_id, pending.item_title, + ) + _record_created_item( + created, actual_category, pending.category_name, vault_log, category_counts, + ) + return True + except Exception as exc: + print( + f" failed to create {pending.category_name} item {pending.item_title!r}: {exc}", + file=sys.stderr, + ) + if import_log is not None: + import_log["failures"].append({ + "vault_title": vault_title, + "vault_id": vault_id, + "item_title": pending.item_title, + "category": pending.category_name, + "error": str(exc), + }) + return False + + +async def prepare_vaults(client, sdk, vault_titles: List[str]) -> Dict[str, tuple[str, bool]]: + """Create missing vaults up front. Returns {title: (vault_id, created)}.""" + existing = await client.vaults.list(sdk.VaultListParams(decrypt_details=True)) + title_to_id = {v.title: v.id for v in existing} + result: Dict[str, tuple[str, bool]] = {} + + for title in vault_titles: + if title in title_to_id: + print(f" vault {title!r} already exists (id={title_to_id[title]}); reusing it") + result[title] = (title_to_id[title], False) + continue + + params = sdk.VaultCreateParams( + title=title, description="Imported from Pleasant Password Server export", + ) + created = await client.vaults.create(params) + print(f" created vault {title!r} (id={created.id})") + result[title] = (created.id, True) + title_to_id[title] = created.id + + return result + + +async def import_vault_entries( + client, + sdk, + vault_id: str, + vault_title: str, + entries: List[dict], + vault_log: Optional[dict], + import_log: Optional[dict], + category_counts: dict, + *, + dry_run: bool, +) -> int: + total_items = 0 + + if dry_run: + for entry in entries: + category_name = classify_entry(entry["raw_fields"], entry["attachments"]) + category_counts[category_name] += 1 + item_title = entry["raw_fields"].get("Title", "(untitled)") + print(f" [dry-run] would create {category_name} item {item_title!r}") + total_items += 1 + return total_items + + batchable: List[_PendingItem] = [] + individual: List[_PendingItem] = [] + + for entry in entries: + category_name = classify_entry(entry["raw_fields"], entry["attachments"]) + category_counts[category_name] += 1 + item_title = entry["raw_fields"].get("Title", "(untitled)") + try: + params = build_item(sdk, category_name, entry, vault_id) + except Exception as exc: + print( + f" failed to build {category_name} item {item_title!r}: {exc}", + file=sys.stderr, + ) + if import_log is not None: + import_log["failures"].append({ + "vault_title": vault_title, + "vault_id": vault_id, + "item_title": item_title, + "category": category_name, + "error": str(exc), + }) + continue + + pending = _PendingItem(entry, category_name, item_title, params) + if needs_individual_create(category_name, params): + individual.append(pending) + else: + batchable.append(pending) + + batch_created = 0 + for chunk in _chunked(batchable, BULK_CREATE_MAX): + try: + resp = await client.items.create_all(vault_id, [p.params for p in chunk]) + except Exception as exc: + print( + f" batch create failed ({len(chunk)} item(s)): {exc}; retrying individually", + file=sys.stderr, + ) + for pending in chunk: + if await _create_one_item( + client, sdk, pending, vault_id, vault_title, + vault_log, import_log, category_counts, + ): + total_items += 1 + continue + + for i, ir in enumerate(resp.individual_responses): + pending = chunk[i] + if ir.error is not None: + if await _create_one_item( + client, sdk, pending, vault_id, vault_title, + vault_log, import_log, category_counts, + ): + total_items += 1 + continue + + _record_created_item( + ir.content, pending.category_name, pending.category_name, + vault_log, category_counts, + ) + batch_created += 1 + total_items += 1 + + if batchable: + print(f" batch created {batch_created}/{len(batchable)} item(s)") + + for pending in individual: + if await _create_one_item( + client, sdk, pending, vault_id, vault_title, + vault_log, import_log, category_counts, + ): + total_items += 1 + + return total_items + + +# -------------------------------------------------------------------------- +# 1Password import +# -------------------------------------------------------------------------- + +def resolve_auth(account_arg: Optional[str] = None): + """Return a service account token or DesktopAuth for Client.authenticate().""" + token = os.environ.get("OP_SERVICE_ACCOUNT_TOKEN") + account = account_arg or os.environ.get("OP_ACCOUNT_NAME") + + if token and account: + print( + "Set either OP_SERVICE_ACCOUNT_TOKEN or OP_ACCOUNT_NAME/--account, not both.", + file=sys.stderr, + ) + sys.exit(1) + if token: + return token + if account: + from onepassword import DesktopAuth + return DesktopAuth(account_name=account) + + print( + "No 1Password credentials found. Set OP_SERVICE_ACCOUNT_TOKEN for a service " + "account, or OP_ACCOUNT_NAME (or --account) for desktop app auth.", + file=sys.stderr, + ) + print( + "Desktop auth requires the 1Password app with " + "Settings → Developer → Integrate with other apps enabled.", + file=sys.stderr, + ) + sys.exit(1) + + +class SdkHandles: + """Small bag of references to the imported SDK names, so build_item() + doesn't need a dozen separate imports passed around.""" + def __init__(self): + from onepassword import ( + AutofillBehavior, Client, DesktopAuth, DocumentCreateParams, ItemCategory, + ItemCreateParams, ItemField, ItemFieldType, ItemSection, VaultCreateParams, + VaultListParams, Website, + ) + + self.Client = Client + self.DesktopAuth = DesktopAuth + self.AutofillBehavior = AutofillBehavior + self.DocumentCreateParams = DocumentCreateParams + self.ItemCategory = ItemCategory + self.ItemCreateParams = ItemCreateParams + self.ItemField = ItemField + self.ItemFieldType = ItemFieldType + self.ItemSection = ItemSection + self.VaultCreateParams = VaultCreateParams + self.VaultListParams = VaultListParams + self.Website = Website + + +def default_log_path(xml_file: str) -> str: + """Default log path in the current working directory, named after the export file.""" + stem = Path(xml_file).name + if stem.lower().endswith(".xml"): + stem = stem[:-4] + return str(Path.cwd() / f"{stem}.pps-import.json") + + +def resolve_log_path(args) -> str: + if args.log_file: + return str(Path(args.log_file).resolve()) + if args.xml_file: + return default_log_path(args.xml_file) + return "" + + +def new_import_log(xml_file: str) -> dict[str, Any]: + return { + "source_xml": str(Path(xml_file).resolve()), + "imported_at": datetime.now(timezone.utc).isoformat(), + "vaults": [], + "failures": [], + } + + +def append_vault_log(import_log: dict[str, Any], *, title: str, vault_id: str, created: bool) -> dict[str, Any]: + vault_log = {"title": title, "id": vault_id, "created": created, "items": []} + import_log["vaults"].append(vault_log) + return vault_log + + +def write_import_log(path: str, import_log: dict[str, Any]) -> None: + with open(path, "w", encoding="utf-8") as f: + json.dump(import_log, f, indent=2) + f.write("\n") + print(f"\nImport log written to {path}") + + +async def delete_import(args): + log_path = resolve_log_path(args) + if not log_path: + print("Provide --log-file or an xml_file to locate the import log.", file=sys.stderr) + sys.exit(1) + if not os.path.isfile(log_path): + print(f"Import log not found: {log_path}", file=sys.stderr) + sys.exit(1) + + with open(log_path, encoding="utf-8") as f: + import_log = json.load(f) + + vaults = import_log.get("vaults") or [] + if not vaults: + print(f"No vaults recorded in {log_path}; nothing to delete.") + return + + created_vaults = [v for v in vaults if v.get("created")] + reused_vaults = [v for v in vaults if not v.get("created")] + reused_items = sum(len(v.get("items") or []) for v in reused_vaults) + + print(f"Import log: {log_path}") + print(f" source: {import_log.get('source_xml', '(unknown)')}") + print(f" imported at: {import_log.get('imported_at', '(unknown)')}") + print(f" vaults to delete: {len(created_vaults)}") + print(f" items to delete from reused vaults: {reused_items}") + + if args.dry_run: + for vault in created_vaults: + print(f" [dry-run] would delete vault {vault['title']!r} (id={vault['id']})") + for vault in reused_vaults: + for item in vault.get("items") or []: + print( + f" [dry-run] would delete item {item['title']!r} " + f"(id={item['id']}) from reused vault {vault['title']!r}" + ) + return + + try: + sdk = SdkHandles() + except ImportError as exc: + print(f"Could not import onepassword-sdk: {exc}", file=sys.stderr) + sys.exit(1) + + auth = resolve_auth(args.account) + auth_label = "service account" if isinstance(auth, str) else f"desktop app ({auth.account_name})" + print(f"Authenticating via {auth_label}...") + client = await sdk.Client.authenticate( + auth=auth, + integration_name="Pleasant Password Server Import Script", + integration_version="1.0.0", + ) + + deleted_vaults = 0 + deleted_items = 0 + for vault in created_vaults: + vault_id = vault.get("id") + if not vault_id or str(vault_id).startswith("DRY-RUN-"): + continue + try: + await client.vaults.delete(vault_id) + print(f" deleted vault {vault['title']!r} (id={vault_id})") + deleted_vaults += 1 + except Exception as exc: + print(f" failed to delete vault {vault['title']!r}: {exc}", file=sys.stderr) + + for vault in reused_vaults: + vault_id = vault.get("id") + if not vault_id or str(vault_id).startswith("DRY-RUN-"): + continue + for item in vault.get("items") or []: + item_id = item.get("id") + if not item_id: + continue + try: + await client.items.delete(vault_id, item_id) + print( + f" deleted item {item['title']!r} (id={item_id}) " + f"from reused vault {vault['title']!r}" + ) + deleted_items += 1 + except Exception as exc: + print( + f" failed to delete item {item['title']!r} from {vault['title']!r}: {exc}", + file=sys.stderr, + ) + + print(f"\nCleanup complete. Deleted {deleted_vaults} vault(s) and {deleted_items} item(s).") + if deleted_vaults == len(created_vaults) and deleted_items == reused_items: + try: + os.remove(log_path) + print(f"Removed import log {log_path}") + except OSError as exc: + print(f"Could not remove import log: {exc}", file=sys.stderr) + + +async def run(args): + vaults_data = parse_pps_xml(args.xml_file) + + total_entries = sum(len(v["entries"]) for v in vaults_data) + print(f"Parsed {len(vaults_data)} vault(s) / {total_entries} entrie(s) from {args.xml_file}") + preview_category_counts = defaultdict(int) + for v in vaults_data: + print(f" - {v['vault_title']!r}: {len(v['entries'])} item(s)") + if args.list_only: + for entry in v["entries"]: + cat = classify_entry(entry["raw_fields"], entry["attachments"]) + preview_category_counts[cat] += 1 + title = entry["raw_fields"].get("Title", "(untitled)") + print(f" [{cat}] {title}") + print() + + if args.list_only: + print("Category breakdown:") + for cat, count in sorted(preview_category_counts.items()): + print(f" {cat}: {count}") + return + + try: + sdk = SdkHandles() + except ImportError as exc: + print(f"Could not import onepassword-sdk: {exc}", file=sys.stderr) + print("Install or upgrade with: pip install -r requirements.txt", + file=sys.stderr) + sys.exit(1) + + client = None + if not args.dry_run: + auth = resolve_auth(args.account) + auth_label = "service account" if isinstance(auth, str) else f"desktop app ({auth.account_name})" + print(f"Authenticating via {auth_label}...") + client = await sdk.Client.authenticate( + auth=auth, + integration_name="Pleasant Password Server Import Script", + integration_version="1.0.0", + ) + + total_items = 0 + category_counts = defaultdict(int) + import_log = None if args.dry_run else new_import_log(args.xml_file) + log_path = resolve_log_path(args) if import_log is not None else "" + + vault_titles = [v["vault_title"] for v in vaults_data] + vault_map: Dict[str, tuple[str, bool]] = {} + if args.dry_run: + vault_map = {title: (f"DRY-RUN-VAULT-ID:{title}", True) for title in vault_titles} + else: + print(f"Preparing {len(vault_titles)} vault(s)...") + vault_map = await prepare_vaults(client, sdk, vault_titles) + + for vault_entry in vaults_data: + title = vault_entry["vault_title"] + entries = vault_entry["entries"] + print(f"Vault: {title} ({len(entries)} item(s))") + + vault_id, vault_created = vault_map[title] + if args.dry_run: + print(f" [dry-run] would create/reuse vault {title!r}") + + vault_log = None + if import_log is not None: + vault_log = append_vault_log( + import_log, title=title, vault_id=vault_id, created=vault_created, + ) + + total_items += await import_vault_entries( + client, sdk, vault_id, title, entries, vault_log, import_log, + category_counts, dry_run=args.dry_run, + ) + + print(f"\nDone. {'Would have processed' if args.dry_run else 'Processed'} " + f"{len(vaults_data)} vault(s) / {total_items} item(s).") + print("Category breakdown:") + for cat, count in sorted(category_counts.items()): + print(f" {cat}: {count}") + + if import_log is not None: + write_import_log(log_path, import_log) + if import_log["failures"]: + print(f" {len(import_log['failures'])} item(s) failed; see log for details.") + + +def main(): + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument( + "xml_file", + nargs="?", + help="Path to the Pleasant Password Server XML export (not required with --delete-import if --log-file is set)", + ) + parser.add_argument("--dry-run", action="store_true", help="Preview actions without calling the 1Password API") + parser.add_argument("--list-only", action="store_true", help="Only parse and print the vault/item summary, no SDK calls at all") + parser.add_argument( + "--account", + metavar="NAME", + help="1Password account name for desktop app auth (overrides OP_ACCOUNT_NAME)", + ) + parser.add_argument( + "--log-file", + metavar="PATH", + help="Import log path (default: ./.pps-import.json in the current directory)", + ) + parser.add_argument( + "--delete-import", + action="store_true", + help="Delete vaults and items recorded in a previous import log instead of importing", + ) + args = parser.parse_args() + + if args.delete_import: + if args.list_only: + parser.error("--delete-import cannot be used with --list-only") + asyncio.run(delete_import(args)) + return + + if not args.xml_file: + parser.error("xml_file is required unless using --delete-import") + if args.list_only and args.dry_run: + parser.error("--list-only cannot be used with --dry-run") + + asyncio.run(run(args)) + + +if __name__ == "__main__": + main() diff --git a/1password/migration/PPS/requirements.txt b/1password/migration/PPS/requirements.txt new file mode 100644 index 0000000..f309049 --- /dev/null +++ b/1password/migration/PPS/requirements.txt @@ -0,0 +1 @@ +onepassword-sdk==0.4.0