diff --git a/CHANGELOG.md b/CHANGELOG.md index 8462587..b671be9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,43 @@ the git history for the fine print. ## Unreleased +- **New: `inventory()` and `python -m httpx_pki inventory` — find the + identities in a directory of certificate exports.** The folder a CA hands + over mixes PKCS#12 bundles, extracted PEM halves, chain bundles, and + issuance artifacts, under extensions that promise nothing. `inventory` + classifies every file by content, pairs private keys with certificates + across files by public key, and reports the constructor call each pairing + amounts to — the work otherwise done by opening files one at a time in an + editor. Where a file holds both halves it is named on its own, rather than + paired with a loose copy of the same key elsewhere in the folder. + + It takes several passwords, not one, because such folders span several; + the report refers to them by position (`password #2`) and never repeats a + value. A file no password opens is reported as `LOCKED` rather than + skipped — silence about a file is the failure mode this exists to remove — + and so is a file that opened in part and kept a key shut, the shape + `openssl pkcs12 -out client.pem` writes. Certificate-only files are + `UNPAIRED`, annotated when they hold an identity's issuer (an alternative + `chain=`/`verify=` source); CSRs and human-readable text dumps are labeled + by which encoded file they describe, via public-key and fingerprint + matching. The same leaf reachable two ways (a `.p12` and its extracted + halves) is presented as one certificate with two routes. Filenames are + treated as untrusted, like the names inside the certificates: control + characters are stripped from the report, and the suggested call quotes the + name as a Python literal. + + Inventory classifies and pairs; it does not audit — `explain()` is the next + step for a source it names — and it deliberately never *builds* a session: + these folders routinely hold several identities and expired renewals, so + choosing one silently is the mistake the report exists to prevent. Top + level only; subdirectories are counted, not descended into. A symlink to a + file is followed and named as it appears here — somebody linked it in on + purpose — while anything that is not a regular file (a device, a socket, a + link pointing nowhere) is named without being read. The CLI + prompts once per still-locked file (skippable), takes repeatable + `--password-env` (no `--password`, same reasoning as `explain`), and exits + non-zero only when nothing loadable was found. + - **Improved: `from_key_pair` names a swapped certificate and private key.** Handing the private key as `certificate=` (or vice versa) used to fail with a generic "could not parse" error, which reads as a broken file. When a diff --git a/README.md b/README.md index 4cc78ea..b5c8110 100644 --- a/README.md +++ b/README.md @@ -82,6 +82,43 @@ PKIClient.from_env() → [Loading certificates](https://httpx-pki.readthedocs.io/en/stable/guide/loading-certificates.html) +## Handed a whole folder? + +A CA rarely sends one file. `inventory` reads the folder, says what each file +actually is, pairs the keys with their certificates, and prints the call each +pairing amounts to: + +```console +$ python -m httpx_pki inventory ./corp-export +``` + +```text +INVENTORY corp-export — 7 files, 2 identities + +IDENTITY svc-client RSA-2048 expires 2027-01-15 + bundle corp.p12 (password #1) + chain corp-issuing-ca.crt + → PKIClient("corp.p12", password=..., chain="corp-issuing-ca.crt") + +IDENTITY svc-client RSA-2048 expires 2027-01-15 + certificate svc-client.pem + private key svc-client.key (encrypted — opened with password #2) + chain corp-issuing-ca.crt + same certificate as corp.p12 + → from_key_pair(certificate="svc-client.pem", private_key="svc-client.key", password=..., chain="corp-issuing-ca.crt") + +LOCKED old-2025.pem — 1 certificate, plus 1 encrypted private key none of the given passwords open + +NOTES cert-details.txt — human-readable dump; fingerprint matches corp.p12 (not loadable) + svc-client.csr — certificate request for the key of corp.p12 (issuance artifact, not loadable) +``` + +Nothing is skipped: a file no password opens is reported as locked, not +dropped. It classifies and pairs — it never builds a session for you, because a +folder like that usually holds more than one answer. + +→ [Taking inventory of a folder](https://httpx-pki.readthedocs.io/en/stable/guide/taking-inventory.html) + ## Async ```python diff --git a/docs/about/how-it-works.md b/docs/about/how-it-works.md index 4723102..cdc58b5 100644 --- a/docs/about/how-it-works.md +++ b/docs/about/how-it-works.md @@ -68,7 +68,8 @@ For anyone reading the source: | `_ssl` | Building the `ssl.SSLContext`, staging, and `verify=` | | `_audit` | Finding trust anchors and chain certificates that cannot do their job | | `_explain` | Laying that out as a report — `explain()` and `client.explain()` | -| `__main__` | `python -m httpx_pki explain` | +| `_inventory` | Classifying a directory of files and pairing them — `inventory()` | +| `__main__` | `python -m httpx_pki explain` and `inventory` | | `_mixin` | Everything the client classes share: constructors, reload, validity | | `_client` | The two public classes, binding the mixin to its httpx base | | `_winstore` / `_keychain` | The OS certificate stores | diff --git a/docs/about/non-goals.md b/docs/about/non-goals.md index d305ad2..a2961f5 100644 --- a/docs/about/non-goals.md +++ b/docs/about/non-goals.md @@ -55,7 +55,9 @@ policy. Beyond that, revocation is out of scope. See ## Fetching anything over the network httpx-pki never makes a request of its own. Reading a certificate does not -cause one, and neither does [`explain()`](../guide/inspecting-a-certificate.md). +cause one, and neither does [`explain()`](../guide/inspecting-a-certificate.md) +or [`inventory()`](../guide/taking-inventory.md) — the latter reads the one +directory you name, top level only, and nothing else. This is a security boundary, not an omission. When a chain is incomplete, `explain()` reports the URL the certificate names for its issuer — its diff --git a/docs/about/security.md b/docs/about/security.md index d2f314b..79045a1 100644 --- a/docs/about/security.md +++ b/docs/about/security.md @@ -57,6 +57,11 @@ That is a deliberate trade, not an oversight. If it is not one you want, reload manually and pass the password each time. See [](../guide/expiry-and-rotation.md#passwords-and-unattended-reloads). +Reports never carry a password value either. +[`inventory()`](../guide/taking-inventory.md) takes several passwords and +refers to them by position — `password #2` — precisely because its output is +meant to be pasted into a ticket or a CI log. + ## `SSLKEYLOGFILE` decrypts your traffic Contexts httpx-pki builds honor the standard `SSLKEYLOGFILE` variable, writing diff --git a/docs/guide/index.md b/docs/guide/index.md index 0939814..e839572 100644 --- a/docs/guide/index.md +++ b/docs/guide/index.md @@ -9,6 +9,8 @@ instead — and if you have an error in hand, Where your credential lives, and how to point httpx-pki at it. +- [](taking-inventory.md) — before you know which file to point at: what a + folder of exports holds, which files pair up, and how to load each pairing - [](loading-certificates.md) — PKCS#12, PEM, separate key and certificate, PKCS#7 chains, and why the file extension never matters - [](windows-store.md) — pulling an exportable certificate out of the Windows diff --git a/docs/guide/inspecting-a-certificate.md b/docs/guide/inspecting-a-certificate.md index a5f5432..23146d6 100644 --- a/docs/guide/inspecting-a-certificate.md +++ b/docs/guide/inspecting-a-certificate.md @@ -256,6 +256,10 @@ For a file you have not written any code for yet: $ python -m httpx_pki explain corp.p12 ``` +(For a whole *folder* you have not written any code for yet, the command is +[`inventory`](taking-inventory.md) — it names the files, and `explain` takes it +from there.) + It takes the same selectors the library does, so a bundle holding several identities can be listed and then inspected: @@ -314,6 +318,8 @@ To make expiry a hard failure on every request instead, use ## Next steps +- [](taking-inventory.md) — the step before this one, when what you have is a + folder rather than a file - [](choosing-a-certificate.md) — inspecting a file that holds several identities - [](expiry-and-rotation.md) — acting on what you find as certificates age diff --git a/docs/guide/taking-inventory.md b/docs/guide/taking-inventory.md new file mode 100644 index 0000000..466506b --- /dev/null +++ b/docs/guide/taking-inventory.md @@ -0,0 +1,208 @@ +# Taking inventory of a folder + +Certificates rarely arrive as one file. What usually lands in your hands is a folder — +a `.p12` next to the PEM halves somebody extracted from it, a chain bundle, a +CSR nobody deleted, a text dump of a certificate that may or may not be one of +the others, and last year's renewal. The extensions promise nothing, several +of the files want passwords, and you do not yet know which two of them go +together. + +`inventory()` reads the folder and tells you: what each file is, which files +pair into a usable identity, and the constructor call each pairing amounts to. + +```console +$ python -m httpx_pki inventory ./corp-export +``` + +```python +import httpx_pki + +print(httpx_pki.inventory("./corp-export")) +``` + +## What the report says + +With no passwords yet, nothing has opened — and the report is already useful: + +```text +INVENTORY corp-export — 7 files, 0 identities + +LOCKED corp.p12 — encrypted PKCS#12; none of the given passwords open it + old-2025.pem — 1 certificate, plus 1 encrypted private key none of the given passwords open + svc-client.key — 1 encrypted private key; none of the given passwords open it + +UNPAIRED corp-issuing-ca.crt — 2 certificates with no matching key here (all CA certificates — possibly a verify= trust bundle) + svc-client.pem — 1 certificate with no matching key here + +NOTES cert-details.txt — human-readable dump; fingerprint matches svc-client.pem (not loadable) + svc-client.csr — certificate request matching nothing here (issuance artifact, not loadable) + +1 subdirectory not inventoried — point inventory() at them directly +``` + +Every file in the folder appears exactly once. Nothing was skipped for being +unrecognizable, unreadable, or shut — a file the report says nothing about is +the failure this exists to remove, so there is no such file. + +Note what it already knows without a single password: `corp-issuing-ca.crt` +holds only CA certificates and is probably a trust bundle; `cert-details.txt` +is prose *about* `svc-client.pem` rather than anything loadable; the CSR is an +issuance artifact. Classification is done on content, never on the extension. + +## Passwords, plural + +A folder accumulated over time spans several passwords — the export password +and the passphrase on the key somebody extracted from it are routinely +different. So `inventory()` takes a list, and tries each against each +encrypted file: + +```python +print(httpx_pki.inventory("./corp-export", passwords=[p12_password, key_password])) +``` + +```text +INVENTORY corp-export — 7 files, 2 identities + +IDENTITY svc-client RSA-2048 expires 2027-01-15 + bundle corp.p12 (password #1) + chain corp-issuing-ca.crt + → PKIClient("corp.p12", password=..., chain="corp-issuing-ca.crt") + +IDENTITY svc-client RSA-2048 expires 2027-01-15 + certificate svc-client.pem + private key svc-client.key (encrypted — opened with password #2) + chain corp-issuing-ca.crt + same certificate as corp.p12 + → from_key_pair(certificate="svc-client.pem", private_key="svc-client.key", password=..., chain="corp-issuing-ca.crt") + +LOCKED old-2025.pem — 1 certificate, plus 1 encrypted private key none of the given passwords open + +NOTES cert-details.txt — human-readable dump; fingerprint matches corp.p12 (not loadable) + svc-client.csr — certificate request for the key of corp.p12 (issuance artifact, not loadable) + +1 subdirectory not inventoried — point inventory() at them directly +``` + +The folder holds **one** certificate reachable **two** ways, and the report +says so rather than presenting two mysteries: the second identity is marked +`same certificate as corp.p12`. Either call works; the `.p12` is one file +instead of two. + +:::{important} +The report names passwords **by position** — `password #2` — and never repeats +a value. That is deliberate. +::: + +`old-2025.pem` stays locked, and that is the last year's renewal nobody could +open. It is still named, still counted, and still on the list of things to ask +somebody about. + +## What each section means + +**`IDENTITY`** — a private key and its certificate, matched by public key, the +same rule the loaders use. One heading line (subject, key type, expiry), the +files it is made of, and the call that loads it. An expired certificate says +`EXPIRED 2026-01-15` in place of `expires`. + +**`LOCKED`** — a file holding key material that none of your passwords opened. +A file that opened *in part* — a PEM whose certificate is readable and whose +key is not, which is what `openssl pkcs12 -out client.pem` writes — appears +here too, because the shut key is the part worth another password. + +**`UNPAIRED`** — a certificate with no matching key in this folder, or a key +with no matching certificate. Not always a mystery: a file holding an +identity's issuer is annotated as an alternative `chain=` or `verify=` source, +and a file of nothing but CA certificates is called out as a probable trust +bundle. + +**`NOTES`** — everything that is not loadable and not a half: CSRs, text +dumps, unrecognizable files, files too large to be certificate material, and +files that could not be read. CSRs and dumps are matched back to the file they +describe, by public key and by fingerprint respectively. + +## From a shell + +```console +$ python -m httpx_pki inventory ./corp-export +$ python -m httpx_pki inventory # the current directory +``` + +Passwords come from the environment, repeatably: + +```console +$ python -m httpx_pki inventory ./corp-export \ + --password-env P12_PASSWORD --password-env KEY_PASSWORD +``` + +Anything still locked after that is prompted for, one file at a time, and each +prompt can be skipped with a blank line. A password typed for one file is +tried against all of them, since a folder's `.p12` and its extracted key +routinely share one. + +There is deliberately no `--password` flag, for the same reason +[`explain`](inspecting-a-certificate.md#from-a-shell) does not have one: an +argument lands in shell history and in every process listing on the machine. + +The command exits non-zero only when the folder yields **nothing loadable**. +Locked and unpaired files are the normal lint of such a folder, not a failure +of the inventory, so a directory with one usable identity and four mysteries +exits `0`. + +## What it will not do + +**It will not build a session.** A folder like this routinely holds several +identities, expired renewals, and a stray trust bundle, so silently choosing +one is precisely the mistake the report exists to prevent. It hands you the +call and lets you make it. + +**It will not descend into subdirectories.** A CA export is flat. Whatever +else a subtree holds, crawling it uninvited is not this function's job — so +subdirectories are counted and named, and you point the tool at them yourself. +Symlinks to files *are* followed, under the name they wear in this folder: +somebody linked it in on purpose. + +**It will not audit.** `inventory()` classifies and pairs. Once it names a +source, [`explain()`](inspecting-a-certificate.md#explaining-a-whole-configuration) +is the tool for what would stop that source working — validity, chain, +trust, key usage, and the problems that only show up on a handshake: + +```console +$ python -m httpx_pki inventory ./corp-export # which file, and how +$ python -m httpx_pki explain corp-export/corp.p12 # and what is wrong with it +``` + +**It will not touch the network**, or anything outside the directory you name. + +## In code + +`inventory()` returns a {class}`~httpx_pki.DirectoryInventory`. `print()` gives the report above; the attributes give the same thing as data: + +```python +report = httpx_pki.inventory("./corp-export", passwords=[p12_password]) + +if not report.usable: + raise SystemExit("nothing loadable in that folder") + +for identity in report.identities: + print(identity.info.common_name, identity.info.not_valid_after) + print(" ", identity.suggestion) + +for entry in report.locked: + print("still need a password for", entry.name) +``` + +`identities` holds {class}`~httpx_pki.InventoryIdentity` objects — `info` is +the usual {class}`~httpx_pki.CertInfo`, and `bundle_file` or the +`certificate_file`/`key_file` pair names what to load. `files` carries every +file the inventory saw as an {class}`~httpx_pki.InventoryEntry`, whatever +became of it; `locked`, `unpaired`, and `notes` are the report's other +sections. + +## Next steps + +- [](loading-certificates.md) — making the call the report suggested +- [](inspecting-a-certificate.md) — `explain()`, for a source the inventory + has named +- [](choosing-a-certificate.md) — when one of those files holds several + identities +- [](server-trust.md) — what to do with the trust bundle it found diff --git a/docs/index.md b/docs/index.md index 8f7328f..53fdec7 100644 --- a/docs/index.md +++ b/docs/index.md @@ -74,6 +74,7 @@ troubleshooting guide/index guide/backends +guide/taking-inventory guide/loading-certificates guide/choosing-a-certificate guide/inspecting-a-certificate diff --git a/docs/quickstart.md b/docs/quickstart.md index f0cc8be..64d15d5 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -68,6 +68,11 @@ your PKI team sent you generally just works. Use the explicit `from_pkcs12` / `from_pem` constructors when you would rather force one interpretation than rely on detection. +Not sure which of those lines applies to what you were sent? If it arrived as a +folder, `python -m httpx_pki inventory ./that-folder` will tell you — it names +every file and prints the call each usable pairing amounts to. See +[](guide/taking-inventory.md). + :::{tip} If a source holds more than one identity — a dual key pair, or a renewed certificate kept alongside the one it replaced — httpx-pki refuses to guess and diff --git a/docs/reference/api.md b/docs/reference/api.md index 012f36d..5d0e34f 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -62,6 +62,31 @@ Inspect what a source holds before mounting anything — see ## Inspection +Two rungs of the same ladder. `inventory()` takes a *directory* and says what +each file is and which files pair into a loadable identity; `explain()` takes +one *source* the inventory named and says what would stop it working. Neither +builds a client. + +`inventory()` reads a folder of exports — see [](../guide/taking-inventory.md). + +```{eval-rst} +.. autofunction:: httpx_pki.inventory +``` + +```{eval-rst} +.. autoclass:: httpx_pki.DirectoryInventory + :members: +``` + +```{eval-rst} +.. autoclass:: httpx_pki.InventoryIdentity + :members: +``` + +```{eval-rst} +.. autoclass:: httpx_pki.InventoryEntry +``` + `explain()` describes a whole configuration — what a source holds, what it would present, what it would trust, and what would stop it working. See [](../guide/inspecting-a-certificate.md#explaining-a-whole-configuration). diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 86ee56c..949d3d5 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -5,21 +5,34 @@ error at all — the request succeeds and the server still treats you as the wro principal — that is the [last group](#it-connects-as-the-wrong-identity). :::{tip} -**Start here if you were handed a certificate and do not know what is in it.** -`explain()` lays out what a source holds, what it would present, what it would -trust, and what would stop it working — without building a client, and without -needing the load to succeed first: +**Start here if you were handed certificates and do not know what is in them.** +Two commands, depending on how much you have narrowed it down. + +A whole folder, and you do not know which file is which — `inventory()` says +what each file is, which files pair into an identity, and how to load each +pairing: + +```console +$ python -m httpx_pki inventory ./corp-export +``` + +One source, and you want to know what would stop it working — `explain()` lays +out what it holds, what it would present, what it would trust, and what is +wrong with it, without building a client and without needing the load to +succeed first: ```console $ python -m httpx_pki explain corp.p12 ``` ```python +print(httpx_pki.inventory("./corp-export")) print(httpx_pki.explain("corp.p12", password="secret")) print(client.explain()) # when you already have a session ``` -See [](guide/inspecting-a-certificate.md#explaining-a-whole-configuration). +See [](guide/taking-inventory.md) and +[](guide/inspecting-a-certificate.md#explaining-a-whole-configuration). ::: ## Find your error @@ -83,8 +96,17 @@ a PKI team hands out contain no private key at all: Those names are conventions, not guarantees — as everywhere else in httpx-pki, [what counts is the bytes](guide/loading-certificates.md#the-extension-does-not-matter). -The reliable way to find out what you have is to try to load it: httpx-pki -inspects the content when a load fails and tells you what it actually found. + +If both halves are somewhere in one folder, +[`inventory()`](guide/taking-inventory.md) reads all of it and says which two +files pair up: + +```console +$ python -m httpx_pki inventory ./corp-export +``` + +And if you have a single file in hand, try to load it: httpx-pki inspects the +content when a load fails and tells you what it actually found. ### "…no private key…" @@ -111,6 +133,10 @@ private key; use it as chain= in from_key_pair or as a verify= CA bundle PKIClient.from_key_pair("client.crt", "client.key") ``` +Not sure which file the key is in — or whether it is even the right key? +`python -m httpx_pki inventory` matches keys to certificates by public key and +prints the `from_key_pair` call for each pair it finds. + **2. You were sent the chain, not your credential.** A `.p7b` from a Windows CA is frequently the *issuing chain* — useful as `chain=` when you present your certificate, or as a `verify=` CA bundle for checking the server, but never a @@ -133,7 +159,9 @@ CertificateLoadError: PKCS#12 data contains no certificate ``` A key with no certificate is equally unusable — you have the proof but not the -claim. The fix is the same: find the other half. +claim. The fix is the same: find the other half — and +[](guide/taking-inventory.md) is how to find it, if it is anywhere in the same +folder. ## The key and certificate do not match @@ -156,6 +184,12 @@ over from a CSR that was superseded. Catching it here is deliberate. Left alone it surfaces much later as an unexplained handshake rejection, with nothing pointing at the file pair. +To find the pairing that *does* work, run +[`inventory()`](guide/taking-inventory.md) over the folder both files came +from. It matches every key against every certificate by public key, so a +superseded key and the certificate it no longer belongs to are reported apart — +each as unpaired, alongside whatever they each really go with. + ## It fails when you make a request The certificate loaded, so the problem is the connection. httpx surfaces these diff --git a/httpx_pki/__init__.py b/httpx_pki/__init__.py index 753a419..0fdd926 100644 --- a/httpx_pki/__init__.py +++ b/httpx_pki/__init__.py @@ -19,6 +19,7 @@ UnsupportedPlatformError, ) from ._explain import TrustAnchor, X509Explanation, explain +from ._inventory import DirectoryInventory, InventoryEntry, InventoryIdentity, inventory from ._keychain import ( MacCert, list_macos_certificates, @@ -40,6 +41,10 @@ "HTTP_BACKEND", "build_ssl_context", "explain", + "inventory", + "DirectoryInventory", + "InventoryIdentity", + "InventoryEntry", "X509Explanation", "Problem", "ChainLink", diff --git a/httpx_pki/__main__.py b/httpx_pki/__main__.py index 760aeb5..f2000a2 100644 --- a/httpx_pki/__main__.py +++ b/httpx_pki/__main__.py @@ -1,13 +1,12 @@ """``python -m httpx_pki`` -- inspect a certificate source from a shell. -The one entry point reachable by somebody who has been handed a ``.p12`` and -has not written any code yet, which is exactly the audience the report is for. -A thin wrapper over :func:`~httpx_pki.explain`: everything it knows comes from -there, and it adds only argument parsing, a password prompt, and an exit code. - -A subcommand from the start, with only one to offer, so that adding another -later (inspecting a *server's* chain, say) does not change how this one is -spelled. +The entry points reachable by somebody who has been handed certificate files +and has not written any code yet, which is exactly the audience the reports +are for. Thin wrappers over :func:`~httpx_pki.explain` (one source: what it +holds and what would stop it working) and :func:`~httpx_pki.inventory` (a whole +directory: what each file is and which files pair up): everything they know +comes from there, and this module adds only argument parsing, password +prompts, and exit codes. """ from __future__ import annotations @@ -19,6 +18,7 @@ from ._exceptions import PKIError from ._explain import explain +from ._inventory import inventory from ._select import selector_from_string, usages_from_string @@ -73,6 +73,37 @@ def _explain(args: argparse.Namespace) -> int: return 1 if report.problems else 0 +def _inventory_passwords(args: argparse.Namespace) -> list[str]: + """The passwords named by the repeated ``--password-env`` flags.""" + passwords = [] + for var in args.password_env: + value = os.environ.get(var) + if value is None: + raise SystemExit(f"environment variable {var} is not set") + passwords.append(value) + return passwords + + +def _inventory(args: argparse.Namespace) -> int: + passwords = _inventory_passwords(args) + report = inventory(args.directory, passwords=passwords or None) + # One prompt per locked file, skippable, then a single re-read: a password + # typed for one file is tried against all of them, since a folder's .p12 + # and its extracted key routinely share a passphrase. + if report.locked and sys.stdin.isatty(): + entered = [] + for item in report.locked: + value = getpass.getpass(f"Password for {item.name} (blank to skip): ") + if value: + entered.append(value) + if entered: + report = inventory(args.directory, passwords=[*passwords, *entered]) + print(report) + # Non-zero only when nothing here is loadable: locked and unpaired files + # are the normal lint of such a folder, not a failure of the inventory. + return 0 if report.usable else 1 + + def main(argv: list[str] | None = None) -> int: """Parse *argv* and run the named subcommand; returns the exit status.""" parser = argparse.ArgumentParser( @@ -143,6 +174,30 @@ def main(argv: list[str] | None = None) -> int: ) explain_parser.set_defaults(func=_explain) + inventory_parser = sub.add_parser( + "inventory", + help="classify a directory of certificate files and pair its identities", + ) + inventory_parser.add_argument( + "directory", + nargs="?", + default=".", + help="the directory to inventory (top-level files only; default: here)", + ) + inventory_parser.add_argument( + "--password-env", + metavar="VAR", + action="append", + default=[], + help=( + "read a password from this environment variable; repeatable, since " + "a folder of exports routinely spans several passwords. There is " + "no --password: it would land in shell history and in every " + "process listing. Files still locked are prompted for, one each" + ), + ) + inventory_parser.set_defaults(func=_inventory) + args = parser.parse_args(argv) try: return int(args.func(args)) diff --git a/httpx_pki/_inventory.py b/httpx_pki/_inventory.py new file mode 100644 index 0000000..b039bb1 --- /dev/null +++ b/httpx_pki/_inventory.py @@ -0,0 +1,881 @@ +"""Finding the usable identities in a directory of certificate exports. + +:func:`inventory` is for the folder a CA hands over: a mix of PKCS#12 bundles, +extracted PEM halves, chain bundles, and issuance artifacts, under extensions +that promise nothing. It classifies every file by content, pairs private keys +with certificates across files (by public key, the way the loaders do), and +reports the constructor call each pairing amounts to -- the work otherwise +done by opening files one at a time in an editor. + +Inventory classifies and pairs; it does not audit. Once it names a source, +:func:`~httpx_pki.explain` is the tool for what would stop that source +working. And it never *builds* anything: a folder like this routinely holds +several identities, expired renewals, and stray trust bundles, so choosing +one silently is exactly the mistake the report exists to prevent. + +Every file the inventory reads lands in the report -- as an identity's part, +or as locked, unpaired, or a note. A file that a password fails to open is +reported as locked rather than skipped: silence about a file is the failure +mode this module exists to remove. +""" + +from __future__ import annotations + +import datetime +import json +import re +from collections.abc import Sequence +from dataclasses import dataclass, field +from pathlib import Path + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.serialization import pkcs7 + +from ._audit import _key_description, _plural +from ._exceptions import CertificateLoadError +from ._material import ( + _PEM_BLOCK, + CertInfo, + Password, + _spki, + certificate_info, + encode_password, +) +from ._pkcs12 import _Loaded, load_bundle + +_WIDTH = 11 # the label column, matching the explain() report + +# Beyond this size a file is not certificate material -- the largest honest +# inhabitant of such a folder is a full trust bundle, well under a megabyte. +_MAX_FILE_SIZE = 10 * 1024 * 1024 + +# A PFX header: SEQUENCE, then version INTEGER 3. The version sits right +# after the outer tag and length, so it appears in the first handful of +# bytes; the sniff keeps "wrong password" (a PKCS#12 nothing opens) apart +# from "not a PKCS#12 at all" without parsing either. +_PFX_VERSION = b"\x02\x01\x03" + +_CSR_LABELS = (b"CERTIFICATE REQUEST", b"NEW CERTIFICATE REQUEST") + + +def _clean(text: str) -> str: + """Strip control characters from an untrusted string. + + Same rule as the explain() report: names inside certificates are + attacker-controlled bytes that a terminal would otherwise act on. The + filenames here get the same treatment -- an unpacked export is no more + trustworthy than the certificates inside it, and a name carrying an + escape sequence would otherwise reach the terminal intact. + """ + return "".join(ch for ch in text if ch.isprintable() or ch == " ") + + +def _literal(name: str) -> str: + """A filename as a Python string literal, for a suggested call. + + Not ``_clean``: the suggestion is meant to be pasted and run, so the name + has to survive whole. ``json.dumps`` escapes the quotes that would end + the literal early and the control characters a terminal would act on, + and its output is a valid Python literal for either. + """ + return json.dumps(name) + + +@dataclass(frozen=True) +class InventoryEntry: + """One file as the inventory classified it. + + ``kind`` is a stable token (``"pkcs12"``, ``"pem"``, ``"certificates"``, + ``"key"``, ``"csr"``, ``"dump"``, ``"locked"``, ``"unknown"``, + ``"unreadable"``, ``"oversized"``); ``summary`` is the prose the report + prints for it. ``password_index`` is the 1-based position of the password + that opened the file, ``None`` when none was needed (or none worked). + + ``"locked"`` is the kind of a file that gave up *nothing*; a file that + opened in part and kept a key shut keeps the kind of what it did give up + and appears under ``locked`` as well, since the shut key is the part + worth another password. + """ + + name: str + kind: str + summary: str + password_index: int | None = None + + +@dataclass(frozen=True) +class InventoryIdentity: # pylint: disable=too-many-instance-attributes + """One presentable identity the directory holds, and how to load it. + + Exactly one of ``bundle_file`` (a self-contained source: a PKCS#12, or a + PEM holding both halves) and the ``certificate_file``/``key_file`` pair + is set. ``suggestion`` is the constructor call the parts amount to, with + ``...`` standing where the password goes -- the report never repeats a + password, it names its position. + """ + + info: CertInfo + key_label: str + bundle_file: str | None = None + certificate_file: str | None = None + key_file: str | None = None + chain_file: str | None = None + password_index: int | None = None + needs_password: bool = False + same_certificate_as: str | None = None + suggestion: str = "" + + def lines(self) -> list[str]: + """The identity as report rows (first row is the heading).""" + now = datetime.datetime.now(datetime.timezone.utc) + validity = ( + f"EXPIRED {self.info.not_valid_after:%Y-%m-%d}" + if self.info.not_valid_after < now + else f"expires {self.info.not_valid_after:%Y-%m-%d}" + ) + name = _clean(self.info.common_name or self.info.distinguished_name) + rows = [f"{name} {self.key_label} {validity}"] + opened = ( + f" (password #{self.password_index})" + if self.password_index is not None + else "" + ) + if self.bundle_file is not None: + rows.append(f" bundle {_clean(self.bundle_file)}{opened}") + else: + rows.append(f" certificate {_clean(self.certificate_file or '')}") + state = ( + f" (encrypted — opened with password #{self.password_index})" + if self.password_index is not None + else " (encrypted)" if self.needs_password else "" + ) + rows.append(f" private key {_clean(self.key_file or '')}{state}") + if self.chain_file is not None: + rows.append(f" chain {_clean(self.chain_file)}") + if self.same_certificate_as is not None: + rows.append( + f" same certificate as {_clean(self.same_certificate_as)}" + ) + rows.append(f" → {self.suggestion}") + return rows + + +@dataclass(frozen=True) +class DirectoryInventory: + """What a directory of certificate files holds, and how to use it. + + Returned by :func:`~httpx_pki.inventory`. ``print()`` it for the laid-out + report; ``repr()`` is the same report, for the same reason + :class:`~httpx_pki.X509Explanation` reads in a REPL. ``files`` carries + every file touched, whatever became of it; the other lists are the + report's sections. + """ + + directory: str + files: list[InventoryEntry] = field(default_factory=list) + identities: list[InventoryIdentity] = field(default_factory=list) + locked: list[InventoryEntry] = field(default_factory=list) + unpaired: list[InventoryEntry] = field(default_factory=list) + notes: list[str] = field(default_factory=list) + skipped_subdirs: int = 0 + + @property + def usable(self) -> bool: + """Whether the directory yields at least one loadable identity.""" + return bool(self.identities) + + def __repr__(self) -> str: + return str(self) + + def __str__(self) -> str: + return "\n".join(self._lines()) + + def _lines(self) -> list[str]: + count = len(self.identities) + found = "1 identity" if count == 1 else f"{count} identities" + head = f"{_clean(self.directory)} — {_plural(len(self.files), 'file')}, {found}" + out = ["INVENTORY".ljust(_WIDTH) + head, ""] + for identity in self.identities: + out += _block("IDENTITY", identity.lines()) + [""] + for label, rows in ( + ("LOCKED", [f.summary for f in self.locked]), + ("UNPAIRED", [f.summary for f in self.unpaired]), + ("NOTES", self.notes), + ): + block = _block(label, rows) + if block: + out += block + [""] + if self.skipped_subdirs: + skipped = ( + "1 subdirectory" + if self.skipped_subdirs == 1 + else f"{self.skipped_subdirs} subdirectories" + ) + out += [f"{skipped} not inventoried — point inventory() at them directly", + "",] + while out and out[-1] == "": + out.pop() + return out + + +def _block(label: str, rows: list[str]) -> list[str]: + """*rows* under a left-hand *label*, which appears on the first row only.""" + if not rows: + return [] + pad = " " * _WIDTH + head = label.ljust(_WIDTH) + return [head + rows[0]] + [pad + row if row else "" for row in rows[1:]] + + +# -- per-file examination ---------------------------------------------------- + + +@dataclass +class _Facts: # pylint: disable=too-many-instance-attributes + """Everything one file contributed, before cross-file assembly.""" + + name: str + kind: str + password_index: int | None = None + keys: list[tuple[bytes, bool]] = field(default_factory=list) # (SPKI, encrypted) + certs: list[x509.Certificate] = field(default_factory=list) + p12_identities: list[_Loaded] = field(default_factory=list) + p12_certs: list[x509.Certificate] = field(default_factory=list) + csr_spkis: list[bytes] = field(default_factory=list) + dump_prints: set[str] = field(default_factory=set) + # Keys that are certainly present and certainly shut: counted separately + # from ``keys`` because a file can hand over some of itself and still be + # holding a key back, and both halves of that have to reach the report. + locked_keys: int = 0 + locked_summary: str | None = None + note: str | None = None + + +def _locked_key_summary(name: str, count: int) -> str: + """The report line for keys a file is holding shut.""" + return ( + f"{_clean(name)} — {_plural(count, 'encrypted private key')}; " + f"none of the given passwords open {'it' if count == 1 else 'them'}" + ) + + +def _try_pem_key( + block: bytes, passwords: list[bytes] +) -> tuple[bytes | None, int | None, bool]: + """(SPKI, password index, encrypted?) for a PEM key block, or all-``None``. + + Encrypted is decided by the no-password attempt: ``TypeError`` is + cryptography saying "there is a key here and it wants a password", which + is exactly the locked/broken distinction the report needs. + """ + try: + key = serialization.load_pem_private_key(block, None) + return _spki(key.public_key()), None, False + except TypeError: + pass + except ValueError: + return None, None, False # present but unreadable: not a password problem + for index, password in enumerate(passwords, 1): + try: + key = serialization.load_pem_private_key(block, password) + return _spki(key.public_key()), index, True + except (ValueError, TypeError): + continue + return None, None, True + + +def _examine_pem( # pylint: disable=too-many-branches + facts: _Facts, data: bytes, passwords: list[bytes] +) -> _Facts: + locked_keys = 0 + for match in _PEM_BLOCK.finditer(data): + label = match.group(1) + block = match.group(0) + if b"PRIVATE KEY" in label: + spki, index, encrypted = _try_pem_key(block, passwords) + if spki is None and encrypted: + locked_keys += 1 + elif spki is not None: + facts.keys.append((spki, encrypted)) + if index is not None: + facts.password_index = index + elif label == b"CERTIFICATE": + try: + facts.certs.append(x509.load_pem_x509_certificate(block)) + except ValueError: + pass # a broken block among good ones; the good ones count + elif label == b"PKCS7": + try: + facts.certs.extend(pkcs7.load_pem_pkcs7_certificates(block)) + except ValueError: + pass + elif label in _CSR_LABELS: + try: + csr = x509.load_pem_x509_csr(block) + except ValueError: + continue + facts.csr_spkis.append( + csr.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + ) + facts.locked_keys = locked_keys + if locked_keys and not facts.keys and not facts.certs: + facts.kind = "locked" + facts.locked_summary = _locked_key_summary(facts.name, locked_keys) + else: + facts.kind = "pem" + return facts + + +def _examine_pkcs12(facts: _Facts, data: bytes, passwords: list[bytes]) -> _Facts: + # None and b"" first: an unprotected PKCS#12 spells "no password" both + # ways depending on the tool that wrote it. + candidates: list[tuple[int | None, bytes | None]] = [(None, None), (None, b"")] + candidates += list(enumerate(passwords, 1)) + for index, password in candidates: + try: + bundle = load_bundle(data, password) + except CertificateLoadError: + continue + facts.kind = "pkcs12" + facts.password_index = index + facts.p12_identities = list(bundle.identities) + facts.p12_certs = list(bundle.certificates) + return facts + facts.kind = "locked" + facts.locked_summary = ( + f"{_clean(facts.name)} — encrypted PKCS#12; " + "none of the given passwords open it" + ) + return facts + + +def _examine_binary( # pylint: disable=too-many-return-statements + facts: _Facts, data: bytes, passwords: list[bytes] +) -> _Facts: + if _PFX_VERSION in data[:8]: + return _examine_pkcs12(facts, data, passwords) + try: + facts.certs.append(x509.load_der_x509_certificate(data)) + facts.kind = "certificates" + return facts + except ValueError: + pass + try: + key = serialization.load_der_private_key(data, None) + facts.keys.append((_spki(key.public_key()), False)) + facts.kind = "key" + return facts + except TypeError: + # Encrypted DER PKCS#8: definitely a key, so the passwords get a try. + for index, password in enumerate(passwords, 1): + try: + key = serialization.load_der_private_key(data, password) + except (ValueError, TypeError): + continue + facts.keys.append((_spki(key.public_key()), True)) + facts.password_index = index + facts.kind = "key" + return facts + facts.kind = "locked" + facts.locked_keys = 1 + facts.locked_summary = _locked_key_summary(facts.name, 1) + return facts + except ValueError: + pass + try: + certs = pkcs7.load_der_pkcs7_certificates(data) + except ValueError: + pass + else: + facts.certs.extend(certs) + facts.kind = "certificates" + return facts + # The PFX sniff is a heuristic; a miss must not turn a real (if oddly + # laid out) PKCS#12 into "unknown", so the full parse gets the last word. + attempted = _examine_pkcs12(facts, data, passwords) + if attempted.kind == "pkcs12": + return attempted + facts.kind = "unknown" + facts.locked_summary = None + facts.note = f"{_clean(facts.name)} — not recognizable certificate material" + return facts + + +# A fingerprint as dumps print one: colon/space-separated byte pairs, or a +# bare hex run the length of a SHA-1 or SHA-256 digest. +_HEX_RUN = re.compile( + r"(?:[0-9A-Fa-f]{2}[:\s-]){15,}[0-9A-Fa-f]{2}|[0-9A-Fa-f]{40,64}" +) + + +def _examine_dump(facts: _Facts, data: bytes) -> _Facts: + """A human-readable certificate dump: not loadable, but labelable. + + The fingerprints such dumps print (NSS, Java, Windows styles all include + one) are extracted so the report can say *which* encoded file the dump + describes -- the one piece of information a text dump is still good for. + """ + facts.kind = "dump" + text = data.decode("utf-8", errors="replace") + for run in _HEX_RUN.findall(text): + normalized = "".join(ch for ch in run if ch.isalnum()).upper() + if len(normalized) in (40, 64): # SHA-1 / SHA-256 + facts.dump_prints.add(normalized) + return facts + + +def _examine(name: str, data: bytes, passwords: list[bytes]) -> _Facts: + facts = _Facts(name=name, kind="unknown") + if b"-----BEGIN" in data: + return _examine_pem(facts, data, passwords) + stripped = data.lstrip() + if stripped.startswith((b"Certificate:", b"X509 Certificate:")): + return _examine_dump(facts, data) + if stripped[:1] == b"\x30": + return _examine_binary(facts, data, passwords) + facts.note = f"{_clean(name)} — not recognizable certificate material" + return facts + + +# -- cross-file assembly ----------------------------------------------------- + + +def _holds(facts: _Facts, spki: bytes) -> bool | None: + """Whether *facts* holds the private key for *spki*, and encrypted. + + ``None`` when the file does not hold it at all -- distinct from ``False``, + which means it holds it and no password was needed. + """ + for candidate, encrypted in facts.keys: + if candidate == spki: + return encrypted + return None + + +def _issuer_in(leaf: x509.Certificate, certs: list[x509.Certificate]) -> bool: + return any(cert.subject == leaf.issuer for cert in certs) + + +def _chain_file_for( + leaf: x509.Certificate, + own_certs: list[x509.Certificate], + cert_only: list[_Facts], +) -> str | None: + """The first certificates-only file that holds *leaf*'s issuer, if the + issuer is not already alongside the leaf in its own source.""" + if _issuer_in(leaf, own_certs) or _is_self_issued(leaf): + return None + for facts in cert_only: + if _issuer_in(leaf, facts.certs): + return facts.name + return None + + +def _is_self_issued(cert: x509.Certificate) -> bool: + return cert.subject == cert.issuer + + +def _all_ca(certs: list[x509.Certificate]) -> bool: + for cert in certs: + try: + constraints = cert.extensions.get_extension_for_class( + x509.BasicConstraints + ).value + # DuplicateExtension included: this decides a hint's phrasing, and a + # malformed certificate must not be able to crash the whole report. + except (x509.ExtensionNotFound, x509.DuplicateExtension): + return False + if not constraints.ca: + return False + return bool(certs) + + +def _file_summary(facts: _Facts) -> str: + """One line saying what the file turned out to be.""" + if facts.locked_summary is not None: + return facts.locked_summary + name = _clean(facts.name) + if facts.kind == "pkcs12": + count = len(facts.p12_identities) + inside = "1 identity" if count == 1 else f"{count} identities" + return f"{name} — PKCS#12, {inside}" + if facts.kind in ("pem", "certificates", "key"): + parts = [] + if facts.keys: + parts.append(_plural(len(facts.keys), "private key")) + if facts.certs: + parts.append(_plural(len(facts.certs), "certificate")) + if facts.csr_spkis: + parts.append(_plural(len(facts.csr_spkis), "certificate request")) + summary = f"{name} — {', '.join(parts) or 'no recognizable blocks'}" + if facts.locked_keys: + # The file gave up part of itself and kept a key shut. Saying only + # what opened would leave the interesting half unmentioned. + summary += ( + f", plus {_plural(facts.locked_keys, 'encrypted private key')} " + "none of the given passwords open" + ) + return summary + if facts.kind == "dump": + return f"{name} — human-readable certificate dump" + return facts.note or name + + +def _normalize_passwords( + passwords: Password | Sequence[Password] | None, +) -> list[bytes]: + if passwords is None: + return [] + if isinstance(passwords, (str, bytes)): + passwords = [passwords] + encoded = [] + for password in passwords: + value = encode_password(password) + if value is not None: + encoded.append(value) + return encoded + + +def inventory( # pylint: disable=too-many-locals,too-many-branches,too-many-statements + directory: str | Path, + passwords: Password | Sequence[Password] | None = None, +) -> DirectoryInventory: + """Classify every file in *directory* and pair the identities it holds. + + *passwords* is one password or several: a folder accumulated over time is + exactly where the PKCS#12 password and an extracted key's passphrase + differ. Each is tried against each encrypted file; the report refers to + them by position (``password #2``) and never repeats a value. A file + nothing opens is reported as locked, not skipped. + + Top-level files only. Subdirectories are counted and named as skipped + rather than descended into -- a CA export is flat, and whatever else a + subtree holds, crawling it uninvited is not this function's call. + + A symlink to a file is followed and inventoried under the name it wears + here: somebody linked it into this folder deliberately, and from where + they are standing the certificate *is* in this folder. Anything that is + not a regular file -- a device, a socket, a link pointing nowhere -- is + named without being read. + """ + root = Path(directory) + if not root.is_dir(): + raise CertificateLoadError(f"{str(directory)!r} is not a directory") + encoded = _normalize_passwords(passwords) + + all_facts: list[_Facts] = [] + files: list[InventoryEntry] = [] + notes: list[str] = [] + skipped_subdirs = 0 + for path in sorted(root.iterdir()): + if path.is_dir(): + skipped_subdirs += 1 + continue + name = path.name + shown = _clean(name) + if not path.is_file(): + # Reading these is what must not happen: a FIFO blocks the read + # forever, a character device never ends it (and reports a size + # of zero, so the ceiling below would not stop it), and a link + # pointing nowhere has nothing behind it. Naming them is the + # whole of the job -- an entry that disappears from the report + # reads as a bug in the tool, and a broken link where a bundle + # should be is exactly what somebody is hunting for. + what = "broken symlink" if path.is_symlink() else "not a regular file" + files.append(InventoryEntry(name, "unreadable", f"{shown} — {what}")) + notes.append(f"{shown} — {what}; not read") + continue + try: + size = path.stat().st_size + if size > _MAX_FILE_SIZE: + files.append( + InventoryEntry( + name, "oversized", f"{shown} — too large to inventory" + ) + ) + notes.append( + f"{shown} — {size // (1024 * 1024)} MiB, too large to be " + "certificate material; not inventoried" + ) + continue + data = path.read_bytes() + except OSError as exc: + files.append(InventoryEntry(name, "unreadable", f"{shown} — {exc}")) + notes.append(f"{shown} — could not be read ({exc})") + continue + all_facts.append(_examine(name, data, encoded)) + + # -- assemble identities --------------------------------------------- + identities: list[InventoryIdentity] = [] + seen_leaves: dict[bytes, str] = {} # leaf DER -> file first offering it + spki_to_source: dict[bytes, str] = {} # identity SPKI -> file, for CSR notes + cert_only = [ + f + for f in all_facts + if f.kind in ("pem", "certificates") and f.certs and not f.keys + ] + + def leaf_note(cert: x509.Certificate, source: str) -> str | None: + der = cert.public_bytes(serialization.Encoding.DER) + if der in seen_leaves: + return seen_leaves[der] + seen_leaves[der] = source + return None + + used_as_chain: set[str] = set() + leaves: list[tuple[x509.Certificate, str]] = [] # leaf, file presenting it + for facts in all_facts: + if facts.kind != "pkcs12": + continue + several = len(facts.p12_identities) > 1 + needs_password = facts.password_index is not None + for loaded in facts.p12_identities: + cert = loaded.identity.certificate + args = [_literal(facts.name)] + if needs_password: + args.append("password=...") + if several: + args.append("identity=httpx_pki.for_mtls") + chain_file = _chain_file_for(cert, facts.p12_certs, cert_only) + if chain_file is not None: + args.append(f"chain={_literal(chain_file)}") + used_as_chain.add(chain_file) + spki_to_source.setdefault(_spki(cert.public_key()), facts.name) + leaves.append((cert, facts.name)) + identities.append( + InventoryIdentity( + info=loaded.identity.info, + key_label=_key_description(cert), + bundle_file=facts.name, + chain_file=chain_file, + password_index=facts.password_index, + needs_password=needs_password, + same_certificate_as=leaf_note(cert, facts.name), + suggestion=f"PKIClient({', '.join(args)})", + ) + ) + + # Keys across files, first file offering a given key wins; every + # certificate sharing its public key is one identity, as the loaders see + # it (a renewal kept alongside its predecessor is two identities). + keys: dict[bytes, tuple[_Facts, bool]] = {} + for facts in all_facts: + for spki, encrypted in facts.keys: + keys.setdefault(spki, (facts, encrypted)) + + matched_keys: set[bytes] = set() + for spki, fallback in keys.items(): + seen_certs: set[bytes] = set() + # Files holding this key first, so that when the same certificate + # sits in two places the self-contained copy is the one reported. + ordered = [f for f in all_facts if _holds(f, spki) is not None] + ordered += [f for f in all_facts if _holds(f, spki) is None] + for cert_facts in ordered: + for cert in cert_facts.certs: + if _spki(cert.public_key()) != spki: + continue + der = cert.public_bytes(serialization.Encoding.DER) + if der in seen_certs: + continue + seen_certs.add(der) + matched_keys.add(spki) + spki_to_source.setdefault(spki, cert_facts.name) + leaves.append((cert, cert_facts.name)) + # The key in the certificate's own file wins over a copy of + # it elsewhere: that file loads on its own, and naming the + # stray copy would suggest pairing two files that need no + # pairing at all. + own = _holds(cert_facts, spki) + key_facts, encrypted = ( + (cert_facts, own) if own is not None else fallback + ) + password_arg = ", password=..." if encrypted else "" + chain_file = _chain_file_for(cert, cert_facts.certs, cert_only) + chain_arg = ( + f", chain={_literal(chain_file)}" + if chain_file is not None + else "" + ) + if chain_file is not None: + used_as_chain.add(chain_file) + if cert_facts is key_facts: + suggestion = ( + f"PKIClient({_literal(key_facts.name)}" + f"{password_arg}{chain_arg})" + ) + certificate_file = None + key_file = None + bundle_file: str | None = key_facts.name + else: + suggestion = ( + f"from_key_pair(certificate={_literal(cert_facts.name)}, " + f"private_key={_literal(key_facts.name)}" + f"{password_arg}{chain_arg})" + ) + certificate_file = cert_facts.name + key_file = key_facts.name + bundle_file = None + identities.append( + InventoryIdentity( + info=certificate_info(cert), + key_label=_key_description(cert), + bundle_file=bundle_file, + certificate_file=certificate_file, + key_file=key_file, + chain_file=chain_file, + password_index=( + key_facts.password_index if encrypted else None + ), + needs_password=encrypted, + same_certificate_as=leaf_note( + cert, bundle_file or cert_facts.name + ), + suggestion=suggestion, + ) + ) + + # -- the files nothing claimed ----------------------------------------- + locked: list[InventoryEntry] = [] + unpaired: list[InventoryEntry] = [] + identity_files: set[str] = set() + for identity in identities: + identity_files.update( + name + for name in ( + identity.bundle_file, + identity.certificate_file, + identity.key_file, + ) + if name is not None + ) + + for facts in all_facts: + entry = InventoryEntry( + facts.name, facts.kind, _file_summary(facts), facts.password_index + ) + files.append(entry) + shown = _clean(facts.name) + if facts.kind == "locked": + locked.append(entry) + continue + if facts.locked_keys: + # Part of the file opened, part of it did not. It belongs in + # LOCKED all the same -- so the CLI offers a prompt for it, and + # so the unpaired verdict below does not call its certificate + # orphaned when the key is sitting right there, shut. + locked.append(entry) + continue + if facts.kind == "dump": + described = _dump_subject(facts, all_facts) + notes.append( + f"{shown} — human-readable dump" + + ( + f"; fingerprint matches {_clean(described)}" + if described + else "; matches nothing here" + ) + + " (not loadable)" + ) + continue + if facts.kind == "unknown": + if facts.note: + notes.append(facts.note) + continue + if facts.csr_spkis and not facts.certs and not facts.keys: + for spki in facts.csr_spkis: + owner = spki_to_source.get(spki) + what = ( + f"for the key of {_clean(owner)}" + if owner + else "matching nothing here" + ) + notes.append( + f"{shown} — certificate request {what} " + "(issuance artifact, not loadable)" + ) + continue + if facts.kind == "pem" and not (facts.certs or facts.keys): + notes.append(f"{shown} — PEM armor with no recognizable blocks") + continue + if facts.name in identity_files or facts.kind == "pkcs12": + continue + if facts.name in used_as_chain: + continue + if facts.certs and not facts.keys: + # An unclaimed certificate file is not always a mystery: holding + # an identity's issuer, it is an alternative chain=/verify= + # source, and saying so beats leaving it as an accusation. + issuer_of = next( + ( + source + for leaf, source in leaves + if _issuer_in(leaf, facts.certs) + ), + None, + ) + if issuer_of is not None: + suffix = ( + f" (holds the issuer of {_clean(issuer_of)} — usable as " + "chain= or verify=)" + ) + elif _all_ca(facts.certs): + suffix = " (all CA certificates — possibly a verify= trust bundle)" + else: + suffix = "" + unpaired.append( + InventoryEntry( + facts.name, + facts.kind, + f"{shown} — {_plural(len(facts.certs), 'certificate')} " + f"with no matching key here{suffix}", + facts.password_index, + ) + ) + elif facts.keys and not any(s in matched_keys for s, _ in facts.keys): + unpaired.append( + InventoryEntry( + facts.name, + facts.kind, + f"{shown} — private key with no matching certificate here", + facts.password_index, + ) + ) + elif facts.keys: + # Spoken for, but by another file that keeps the same key next to + # its certificate. A spare copy is worth naming rather than + # passing over in silence. + owner = next( + (spki_to_source[s] for s, _ in facts.keys if s in spki_to_source), + None, + ) + where = f", already paired in {_clean(owner)}" if owner else "" + notes.append(f"{shown} — a second copy of a private key{where}") + + return DirectoryInventory( + directory=str(directory), + files=files, + identities=identities, + locked=locked, + unpaired=unpaired, + notes=notes, + skipped_subdirs=skipped_subdirs, + ) + + +def _dump_subject(facts: _Facts, all_facts: list[_Facts]) -> str | None: + """The file whose certificate a text dump's fingerprint names, if any.""" + if not facts.dump_prints: + return None + for other in all_facts: + for cert in [*other.certs, *other.p12_certs]: + prints = { + cert.fingerprint(hashes.SHA1()).hex().upper(), + cert.fingerprint(hashes.SHA256()).hex().upper(), + } + if prints & facts.dump_prints: + return other.name + return None diff --git a/tests/test_inventory.py b/tests/test_inventory.py new file mode 100644 index 0000000..3236780 --- /dev/null +++ b/tests/test_inventory.py @@ -0,0 +1,720 @@ +"""Tests for inventory(): directory classification, pairing, and the CLI.""" + +from __future__ import annotations + +import ast +import base64 +import getpass +import os +import sys +import textwrap +from pathlib import Path + +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization + +from httpx_pki import CertificateLoadError, _inventory, inventory +from httpx_pki.__main__ import main +from httpx_pki.testing import make_client_cert +from tests.conftest import P12_PASSWORD, Signed + +KEY_PASSWORD = b"keypw" + +DER = serialization.Encoding.DER +PEM = serialization.Encoding.PEM + + +def _encrypted_key(signed: Signed) -> bytes: + return signed.key.private_bytes( + PEM, + serialization.PrivateFormat.PKCS8, + serialization.BestAvailableEncryption(KEY_PASSWORD), + ) + + +def _der_key(signed: Signed, password: bytes | None = None) -> bytes: + encryption: serialization.KeySerializationEncryption = ( + serialization.NoEncryption() + if password is None + else serialization.BestAvailableEncryption(password) + ) + return signed.key.private_bytes( + DER, serialization.PrivateFormat.PKCS8, encryption + ) + + +def _write(directory: Path, **files: bytes) -> Path: + for name, data in files.items(): + (directory / name.replace("_", ".")).write_bytes(data) + return directory + + +# -- PKCS#7 -------------------------------------------------------------------- +# `cryptography` reads certificate-only PKCS#7 but will not write it, and a +# .p7b is exactly what a Windows or Java export drops in such a folder. The +# structure is small enough to encode by hand: a SignedData carrying nothing +# but certificates, which is all a chain bundle ever is. + +_SIGNED_DATA_OID = bytes([0x2A, 0x86, 0x48, 0x86, 0xF7, 0x0D, 0x01, 0x07, 0x02]) +_DATA_OID = bytes([0x2A, 0x86, 0x48, 0x86, 0xF7, 0x0D, 0x01, 0x07, 0x01]) + + +def _tlv(tag: int, payload: bytes) -> bytes: + """One DER tag-length-value, with the length in short or long form.""" + if len(payload) < 0x80: + return bytes([tag, len(payload)]) + payload + width = (len(payload).bit_length() + 7) // 8 + return bytes([tag, 0x80 | width]) + len(payload).to_bytes(width, "big") + payload + + +def _pkcs7(*certs: x509.Certificate) -> bytes: + """*certs* as a certificate-only PKCS#7 blob, DER.""" + signed_data = _tlv( + 0x30, + _tlv(0x02, b"\x01") # version + + _tlv(0x31, b"") # digestAlgorithms: none + + _tlv(0x30, _tlv(0x06, _DATA_OID)) # encapContentInfo: data + + _tlv(0xA0, b"".join(cert.public_bytes(DER) for cert in certs)) + + _tlv(0x31, b""), # signerInfos: none + ) + return _tlv(0x30, _tlv(0x06, _SIGNED_DATA_OID) + _tlv(0xA0, signed_data)) + + +def _armor(label: str, der: bytes) -> bytes: + """*der* wrapped in a PEM block, the way a tool that emits text would.""" + body = "\n".join(textwrap.wrap(base64.b64encode(der).decode(), 64)) + return f"-----BEGIN {label}-----\n{body}\n-----END {label}-----\n".encode() + + +def test_pair_across_files(client: Signed, ca: Signed, tmp_path: Path) -> None: + # The motivating case: cert in one file, encrypted key in another, chain + # in a third, none with a helpful extension. + _write( + tmp_path, + client_pem=client.cert_pem, + client_ukey=_encrypted_key(client), + chain_crt=ca.cert_pem, + ) + report = inventory(tmp_path, passwords=["wrong-first", KEY_PASSWORD]) + assert report.usable + [identity] = report.identities + assert identity.certificate_file == "client.pem" + assert identity.key_file == "client.ukey" + assert identity.chain_file == "chain.crt" + assert identity.password_index == 2 + assert 'from_key_pair(certificate="client.pem"' in identity.suggestion + assert 'private_key="client.ukey"' in identity.suggestion + assert "password=..." in identity.suggestion + assert 'chain="chain.crt"' in identity.suggestion + # The chain file is claimed by the identity, not left as a mystery. + assert not report.unpaired + + +def test_single_file_bundle_suggests_single_source( + client: Signed, tmp_path: Path +) -> None: + _write(tmp_path, bundle_pem=client.key_pem + client.cert_pem) + report = inventory(tmp_path) + [identity] = report.identities + assert identity.bundle_file == "bundle.pem" + assert identity.suggestion == 'PKIClient("bundle.pem")' + + +def test_pkcs12_identity(client_p12: bytes, tmp_path: Path) -> None: + _write(tmp_path, export_p12=client_p12) + report = inventory(tmp_path, passwords=[P12_PASSWORD]) + [identity] = report.identities + assert identity.bundle_file == "export.p12" + assert identity.password_index == 1 + assert identity.suggestion == 'PKIClient("export.p12", password=...)' + + +def test_locked_files_reported_not_skipped( + client: Signed, client_p12: bytes, tmp_path: Path +) -> None: + # The design rule: a file a password fails to open lands in LOCKED, + # because silently dropping it is the notepad failure mode again. + _write(tmp_path, legacy_p12=client_p12, old_ukey=_encrypted_key(client)) + report = inventory(tmp_path, passwords=["wrong"]) + assert not report.usable + assert {f.name for f in report.locked} == {"legacy.p12", "old.ukey"} + assert all("none of the given passwords" in f.summary for f in report.locked) + # Every file is accounted for somewhere. + assert {f.name for f in report.files} == {"legacy.p12", "old.ukey"} + + +def test_passwords_never_appear_in_report( + client: Signed, client_p12: bytes, tmp_path: Path +) -> None: + # The report names password positions, never values: it will be pasted + # into tickets and terminal logs. + _write( + tmp_path, + export_p12=client_p12, + client_pem=client.cert_pem, + client_ukey=_encrypted_key(client), + ) + text = str(inventory(tmp_path, passwords=[P12_PASSWORD, KEY_PASSWORD])) + assert P12_PASSWORD not in text + assert KEY_PASSWORD.decode() not in text + assert "password #1" in text and "password #2" in text + + +def test_same_certificate_grouped( + client: Signed, client_p12: bytes, tmp_path: Path +) -> None: + # The same leaf reachable as a PKCS#12 and as extracted halves is one + # certificate with two routes, and the report must say so rather than + # present two mystery identities. + _write( + tmp_path, + export_p12=client_p12, + client_pem=client.cert_pem, + client_key=client.key_pem, + ) + report = inventory(tmp_path, passwords=[P12_PASSWORD]) + assert len(report.identities) == 2 + pair = next(i for i in report.identities if i.key_file == "client.key") + assert pair.same_certificate_as == "export.p12" + + +def test_partly_locked_file_lands_in_locked(client: Signed, tmp_path: Path) -> None: + # The layout `openssl pkcs12 -out client.pem` writes: both halves in one + # file, the key encrypted. Reporting only the certificate would call it an + # orphan and never offer the password prompt that opens it. + _write(tmp_path, bundle_pem=_encrypted_key(client) + client.cert_pem) + report = inventory(tmp_path, passwords=["wrong"]) + assert not report.usable + [entry] = report.locked + assert entry.name == "bundle.pem" + assert "1 certificate" in entry.summary + assert "encrypted private key" in entry.summary + assert not report.unpaired # not an orphaned certificate: its key is right there + # With the password it is one self-contained source. + opened = inventory(tmp_path, passwords=[KEY_PASSWORD]) + assert not opened.locked + [identity] = opened.identities + assert identity.suggestion == 'PKIClient("bundle.pem", password=...)' + + +def test_self_contained_file_beats_a_stray_key_copy( + client: Signed, tmp_path: Path +) -> None: + # A loose copy of the key sorts first, but the file holding both halves + # loads on its own -- suggesting the two-file call for it would be a + # pairing nobody needs to make. + _write( + tmp_path, + a_copy_key=client.key_pem, + bundle_pem=client.key_pem + client.cert_pem, + ) + report = inventory(tmp_path) + [identity] = report.identities + assert identity.bundle_file == "bundle.pem" + assert identity.suggestion == 'PKIClient("bundle.pem")' + # The spare copy is named rather than passed over. + assert any("a.copy.key" in note and "second copy" in note for note in report.notes) + + +@pytest.mark.skipif(sys.platform == "win32", reason="filename is illegal on NTFS") +def test_hostile_filename_is_neutralized(client: Signed, tmp_path: Path) -> None: + # A filename out of somebody else's archive is as untrusted as the + # certificates inside it: the escape must not reach the terminal, and the + # quote must not end the suggestion's string literal early. + (tmp_path / 'we"ird\x1b[31m.pem').write_bytes(client.key_pem + client.cert_pem) + [identity] = inventory(tmp_path).identities + text = str(inventory(tmp_path)) + assert "\x1b" not in text + assert identity.suggestion == 'PKIClient("we\\"ird\\u001b[31m.pem")' + # The suggestion is still the real filename, so it still works. + assert ast.literal_eval( + identity.suggestion.removeprefix("PKIClient(").removesuffix(")") + ) == 'we"ird\x1b[31m.pem' + + +def test_unpaired_cert_and_key(client: Signed, ca: Signed, tmp_path: Path) -> None: + other = Signed(ca.key, ca.cert) # a cert file with no key alongside + _write(tmp_path, stray_crt=other.cert_pem, stray_key=client.key_pem) + report = inventory(tmp_path) + assert not report.identities + summaries = {f.name: f.summary for f in report.unpaired} + assert "no matching key here" in summaries["stray.crt"] + assert "no matching certificate" in summaries["stray.key"] + + +def test_unpaired_issuer_named(client: Signed, ca: Signed, tmp_path: Path) -> None: + # A cert-only file holding an identity's issuer is an alternative + # chain=/verify= source, not a mystery -- but only when it was not + # already claimed as the suggested chain (a second copy here). + _write( + tmp_path, + bundle_pem=client.key_pem + client.cert_pem + ca.cert_pem, + issuer_pem=ca.cert_pem, + ) + report = inventory(tmp_path) + [entry] = report.unpaired + assert entry.name == "issuer.pem" + assert "holds the issuer of bundle.pem" in entry.summary + + +def test_csr_note_names_the_key(client: Signed, tmp_path: Path) -> None: + csr = ( + x509.CertificateSigningRequestBuilder() + .subject_name(client.cert.subject) + .sign(client.key, hashes.SHA256()) + .public_bytes(serialization.Encoding.PEM) + ) + # The legacy Windows/keytool label must classify the same way. + legacy = csr.replace(b"CERTIFICATE REQUEST", b"NEW CERTIFICATE REQUEST") + _write( + tmp_path, + bundle_pem=client.key_pem + client.cert_pem, + request_csr=legacy, + ) + report = inventory(tmp_path) + assert any( + "request.csr" in note and "for the key of bundle.pem" in note + for note in report.notes + ) + + +def test_dump_note_matches_fingerprint(client: Signed, tmp_path: Path) -> None: + fingerprint = client.cert.fingerprint(hashes.SHA256()).hex(":").upper() + dump = ( + "Certificate:\n Data:\n" + f" Fingerprint (SHA-256):\n {fingerprint}\n" + ) + _write( + tmp_path, + bundle_pem=client.key_pem + client.cert_pem, + info_txt=dump.encode(), + ) + report = inventory(tmp_path) + assert any( + "info.txt" in note and "matches bundle.pem" in note + for note in report.notes + ) + + +def test_subdirectories_counted_not_descended( + client: Signed, tmp_path: Path +) -> None: + (tmp_path / "nested").mkdir() + (tmp_path / "nested" / "bundle.pem").write_bytes( + client.key_pem + client.cert_pem + ) + report = inventory(tmp_path) + assert report.skipped_subdirs == 1 + assert not report.identities + + +@pytest.mark.skipif(sys.platform == "win32", reason="needs POSIX file types") +def test_irregular_entries_named_not_read(client: Signed, tmp_path: Path) -> None: + # A FIFO would block the read forever and a broken link has nothing + # behind it, so neither is opened -- but neither may vanish either: a + # name in `ls` that the report does not mention reads as a broken tool. + _write(tmp_path, bundle_pem=client.key_pem + client.cert_pem) + os.mkfifo(tmp_path / "pipe.pem") + (tmp_path / "gone.p12").symlink_to(tmp_path / "not-there.p12") + report = inventory(tmp_path) + assert {f.name for f in report.files} == {"bundle.pem", "gone.p12", "pipe.pem"} + summaries = {f.name: f.summary for f in report.files} + assert "broken symlink" in summaries["gone.p12"] + assert "not a regular file" in summaries["pipe.pem"] + assert sum("not read" in note for note in report.notes) == 2 + # The real file is still inventoried as usual. + [identity] = report.identities + assert identity.bundle_file == "bundle.pem" + + +@pytest.mark.skipif(sys.platform == "win32", reason="needs POSIX symlinks") +def test_symlinked_file_is_followed(client: Signed, tmp_path: Path) -> None: + # Somebody linked it in on purpose: from where they stand the certificate + # is in this folder, and the report names it as they see it. + elsewhere = tmp_path / "elsewhere" + elsewhere.mkdir() + (elsewhere / "real.pem").write_bytes(client.key_pem + client.cert_pem) + (tmp_path / "linked.pem").symlink_to(elsewhere / "real.pem") + report = inventory(tmp_path) + [identity] = report.identities + assert identity.bundle_file == "linked.pem" + assert identity.suggestion == 'PKIClient("linked.pem")' + assert report.skipped_subdirs == 1 # the directory it points into + + +def test_garbage_and_empty(tmp_path: Path) -> None: + _write(tmp_path, junk_bin=b"\x00\x01garbage") + report = inventory(tmp_path) + assert not report.usable + assert any("junk.bin" in note for note in report.notes) + with pytest.raises(CertificateLoadError, match="not a directory"): + inventory(tmp_path / "missing") + + +def test_cli_inventory_exit_codes( + client: Signed, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + _write(tmp_path, client_pem=client.cert_pem, client_ukey=_encrypted_key(client)) + monkeypatch.setenv("INVENTORY_PW", KEY_PASSWORD.decode()) + assert main(["inventory", str(tmp_path), "--password-env", "INVENTORY_PW"]) == 0 + assert "from_key_pair" in capsys.readouterr().out + # Nothing loadable: non-zero, but still a report rather than an error. + empty = tmp_path / "empty" + empty.mkdir() + assert main(["inventory", str(empty)]) == 1 + assert "0 identities" in capsys.readouterr().out + + +def test_cli_prompts_for_a_locked_file( + client: Signed, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + # A file holding a key back is offered a prompt, whatever else it gave up. + _write(tmp_path, bundle_pem=_encrypted_key(client) + client.cert_pem) + monkeypatch.setattr(sys.stdin, "isatty", lambda: True) + monkeypatch.setattr(getpass, "getpass", lambda prompt: KEY_PASSWORD.decode()) + assert main(["inventory", str(tmp_path)]) == 0 + assert 'PKIClient("bundle.pem", password=...)' in capsys.readouterr().out + + +# -- DER: what a Windows or Java export drops in the folder -------------------- + + +def test_der_certificate_and_key_pair(client: Signed, tmp_path: Path) -> None: + # No PEM armor anywhere: both halves are raw DER under extensions that + # say nothing, which is the ordinary shape of a Windows export. + _write( + tmp_path, + client_cer=client.cert.public_bytes(DER), + client_der=_der_key(client), + ) + [identity] = inventory(tmp_path).identities + assert identity.certificate_file == "client.cer" + assert identity.key_file == "client.der" + assert identity.suggestion == ( + 'from_key_pair(certificate="client.cer", private_key="client.der")' + ) + + +def test_der_encrypted_key_opens_and_locks(client: Signed, tmp_path: Path) -> None: + _write( + tmp_path, + client_cer=client.cert.public_bytes(DER), + client_der=_der_key(client, KEY_PASSWORD), + ) + [identity] = inventory(tmp_path, passwords=["wrong", KEY_PASSWORD]).identities + assert identity.needs_password + assert identity.password_index == 2 + assert "password=..." in identity.suggestion + # An encrypted DER key nothing opens is locked, not mistaken for garbage. + locked = inventory(tmp_path, passwords=["wrong"]) + assert not locked.identities + assert [f.name for f in locked.locked] == ["client.der"] + assert "encrypted private key" in locked.locked[0].summary + + +def test_der_pkcs7_is_read_as_certificates(ca: Signed, tmp_path: Path) -> None: + _write(tmp_path, chain_p7b=_pkcs7(ca.cert)) + [entry] = inventory(tmp_path).unpaired + assert entry.name == "chain.p7b" + assert entry.kind == "certificates" + assert "all CA certificates" in entry.summary + + +def test_pem_pkcs7_is_read_as_certificates(ca: Signed, tmp_path: Path) -> None: + _write(tmp_path, chain_p7c=_armor("PKCS7", _pkcs7(ca.cert))) + [entry] = inventory(tmp_path).unpaired + assert entry.name == "chain.p7c" + assert "1 certificate" in entry.summary + + +def test_der_garbage_is_named_unknown(tmp_path: Path) -> None: + # Starts like DER and is nothing: every parse must fail without the file + # being dropped or blamed on a password. + _write(tmp_path, blob_der=b"\x30\x82\x00\x05hello") + report = inventory(tmp_path, passwords=["irrelevant"]) + assert [f.kind for f in report.files] == ["unknown"] + assert not report.locked + assert any( + "blob.der" in note and "not recognizable" in note for note in report.notes + ) + + +def test_pkcs12_found_when_the_header_sniff_misses( + client_p12: bytes, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # The PFX sniff is a shortcut, not the authority: with it defeated, the + # full parse still has to recognize the bundle. + monkeypatch.setattr(_inventory, "_PFX_VERSION", b"\xff\xff\xff") + _write(tmp_path, export_p12=client_p12) + [identity] = inventory(tmp_path, passwords=[P12_PASSWORD]).identities + assert identity.bundle_file == "export.p12" + + +# -- PEM blocks that do not parse ---------------------------------------------- + + +def test_broken_blocks_do_not_cost_the_good_ones( + client: Signed, tmp_path: Path +) -> None: + # Armor around nonsense, of every label the scan knows. Each is skipped + # on its own; the usable pair in the same file still comes out. + rubbish = base64.b64encode(b"not a key, not a certificate, not a request") + broken = b"".join( + b"-----BEGIN " + label + b"-----\n" + + rubbish + + b"\n-----END " + label + b"-----\n" + for label in (b"PRIVATE KEY", b"CERTIFICATE", b"PKCS7", b"CERTIFICATE REQUEST") + ) + _write(tmp_path, mixed_pem=broken + client.key_pem + client.cert_pem) + report = inventory(tmp_path) + [identity] = report.identities + assert identity.bundle_file == "mixed.pem" + assert not report.locked # unreadable is not the same as password-protected + + +def test_pem_armor_with_no_known_blocks(tmp_path: Path) -> None: + _write(tmp_path, params_pem=_armor("DH PARAMETERS", b"\x30\x03\x02\x01\x00")) + report = inventory(tmp_path) + assert not report.usable + assert any( + "params.pem" in note and "no recognizable blocks" in note + for note in report.notes + ) + + +# -- pairing and de-duplication ------------------------------------------------ + + +def test_two_identities_in_one_file(ca: Signed, tmp_path: Path) -> None: + # A file somebody built with `cat`: two whole identities in one blob. Each + # key finds its own certificate rather than the first one in the file. + first = make_client_cert("first", ca=None) + second = make_client_cert("second", ca=None) + _write( + tmp_path, + both_pem=first.key_pem + first.cert_pem + second.key_pem + second.cert_pem, + ) + report = inventory(tmp_path) + assert len(report.identities) == 2 + assert {i.info.common_name for i in report.identities} == {"first", "second"} + assert all(i.bundle_file == "both.pem" for i in report.identities) + + +def test_the_same_certificate_in_two_files_is_one_identity( + client: Signed, tmp_path: Path +) -> None: + _write(tmp_path, a_pem=client.key_pem + client.cert_pem, b_pem=client.cert_pem) + report = inventory(tmp_path) + [identity] = report.identities + assert identity.bundle_file == "a.pem" + # The second copy is a certificate with no key of its own, and nothing + # more is claimed about it: it neither issues anything nor is a CA. + [entry] = report.unpaired + assert entry.name == "b.pem" + assert entry.summary.endswith("with no matching key here") + + +def test_a_ca_false_certificate_is_not_a_trust_bundle( + ca: Signed, tmp_path: Path +) -> None: + leaf = ( + x509.CertificateBuilder() + .subject_name(x509.Name([x509.NameAttribute(x509.NameOID.COMMON_NAME, "leaf")])) + .issuer_name(ca.cert.subject) + .public_key(ca.key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(ca.cert.not_valid_before_utc) + .not_valid_after(ca.cert.not_valid_after_utc) + .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) + .sign(ca.key, hashes.SHA256()) + ) + _write(tmp_path, stray_pem=leaf.public_bytes(PEM)) + [entry] = inventory(tmp_path).unpaired + assert "trust bundle" not in entry.summary + + +def test_pkcs12_with_two_identities_names_the_selector( + dual_p12: bytes, tmp_path: Path +) -> None: + # Two identities in one bundle: loading it needs a choice, and the + # suggested call has to say so rather than look like a one-liner. + _write(tmp_path, dual_p12=dual_p12) + report = inventory(tmp_path, passwords=[P12_PASSWORD]) + assert len(report.identities) == 2 + assert all( + "identity=httpx_pki.for_mtls" in identity.suggestion + for identity in report.identities + ) + + +def test_pkcs12_names_a_chain_file( + client_p12: bytes, ca: Signed, tmp_path: Path +) -> None: + # The bundle holds no issuer, but the folder does: the report pairs them + # and the issuer file stops being a mystery. + _write(tmp_path, export_p12=client_p12, issuer_pem=ca.cert_pem) + report = inventory(tmp_path, passwords=[P12_PASSWORD]) + [identity] = report.identities + assert identity.chain_file == "issuer.pem" + assert 'chain="issuer.pem"' in identity.suggestion + assert not report.unpaired + + +# -- passwords, and files that cannot be read ---------------------------------- + + +def test_a_single_password_need_not_be_a_list( + client_p12: bytes, tmp_path: Path +) -> None: + _write(tmp_path, export_p12=client_p12) + assert inventory(tmp_path, passwords=P12_PASSWORD).usable + # A None among several is dropped rather than counted as a position. + report = inventory(tmp_path, passwords=[None, P12_PASSWORD]) + assert report.identities[0].password_index == 1 + + +def test_oversized_file_is_named_not_read(tmp_path: Path) -> None: + with open(tmp_path / "huge.pem", "wb") as handle: + handle.truncate(10 * 1024 * 1024 + 1) + report = inventory(tmp_path) + assert [f.kind for f in report.files] == ["oversized"] + assert any("huge.pem" in note and "too large" in note for note in report.notes) + + +@pytest.mark.skipif( + sys.platform == "win32" or os.geteuid() == 0, + reason="needs POSIX permissions and a user that they apply to", +) +def test_unreadable_file_is_named(client: Signed, tmp_path: Path) -> None: + path = tmp_path / "no-access.pem" + path.write_bytes(client.cert_pem) + path.chmod(0) + report = inventory(tmp_path) + assert [f.kind for f in report.files] == ["unreadable"] + assert any("could not be read" in note for note in report.notes) + + +# -- text dumps ---------------------------------------------------------------- + + +def test_dump_without_a_fingerprint(tmp_path: Path) -> None: + # The certutil spelling, and nothing in it to match on. + _write(tmp_path, info_txt=b"X509 Certificate:\nVersion: 3\nSerial Number: 01\n") + report = inventory(tmp_path) + assert [f.kind for f in report.files] == ["dump"] + assert any("matches nothing here" in note for note in report.notes) + + +def test_dump_serial_number_is_not_mistaken_for_a_fingerprint( + client: Signed, tmp_path: Path +) -> None: + # Dumps print serial numbers in the same colon-separated byte pairs as + # fingerprints. Only a digest-length run may be matched on -- claiming the + # wrong file is worse than claiming none. + serial = ":".join(f"{byte:02X}" for byte in range(17)) # 34 hex characters + print_ = client.cert.fingerprint(hashes.SHA256()).hex(":").upper() + _write( + tmp_path, + bundle_pem=client.key_pem + client.cert_pem, + info_txt=f"Certificate:\n Serial: {serial}\n SHA256: {print_}\n".encode(), + ) + report = inventory(tmp_path) + assert any( + "info.txt" in note and "matches bundle.pem" in note for note in report.notes + ) + + +def test_dump_naming_a_certificate_that_is_not_here( + client: Signed, ca: Signed, tmp_path: Path +) -> None: + absent = ca.cert.fingerprint(hashes.SHA256()).hex(":").upper() + _write( + tmp_path, + bundle_pem=client.key_pem + client.cert_pem, + info_txt=f"Certificate:\n SHA256 Fingerprint={absent}\n".encode(), + ) + report = inventory(tmp_path) + assert any( + "info.txt" in note and "matches nothing here" in note for note in report.notes + ) + + +# -- the laid-out report ------------------------------------------------------- + + +def test_report_renders_every_section( + client: Signed, + client_p12: bytes, + ca: Signed, + server_cert: Signed, + tmp_path: Path, +) -> None: + _write( + tmp_path, + export_p12=client_p12, + issuer_pem=ca.cert_pem, + old_ukey=_encrypted_key(client), + stray_crt=server_cert.cert_pem, + junk_bin=b"\x00\x01garbage", + ) + (tmp_path / "archive").mkdir() + report = inventory(tmp_path, passwords=[P12_PASSWORD]) + text = str(report) + assert repr(report) == text # the REPL shows the report, not a dataclass dump + assert "chain issuer.pem" in text + assert "LOCKED" in text and "UNPAIRED" in text and "NOTES" in text + assert "1 subdirectory not inventoried" in text + (tmp_path / "archive-2019").mkdir() + assert "2 subdirectories not inventoried" in str( + inventory(tmp_path, passwords=[P12_PASSWORD]) + ) + + +def test_expired_identity_is_labeled(tmp_path: Path) -> None: + # The renewal nobody deleted: still loadable, and the report says plainly + # that presenting it is pointless. + stale = make_client_cert("old-client", ca=None, expired=True) + _write(tmp_path, old_pem=stale.key_pem + stale.cert_pem) + text = str(inventory(tmp_path)) + assert "EXPIRED" in text + assert "expires" not in text + + +# -- the command line ---------------------------------------------------------- + + +def test_cli_reports_an_unset_password_env( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("NO_SUCH_INVENTORY_PW", raising=False) + with pytest.raises(SystemExit, match="NO_SUCH_INVENTORY_PW is not set"): + main(["inventory", str(tmp_path), "--password-env", "NO_SUCH_INVENTORY_PW"]) + + +def test_cli_prompt_can_be_skipped( + client: Signed, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + # Blank at the prompt means "I do not have it": the file stays locked and + # the report is still printed. + _write(tmp_path, old_ukey=_encrypted_key(client)) + monkeypatch.setattr(sys.stdin, "isatty", lambda: True) + monkeypatch.setattr(getpass, "getpass", lambda prompt: "") + assert main(["inventory", str(tmp_path)]) == 1 + assert "LOCKED" in capsys.readouterr().out + + +def test_cli_error_on_a_missing_directory( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + assert main(["inventory", str(tmp_path / "nowhere")]) == 2 + assert "error:" in capsys.readouterr().err