coulomb/net-kingdom
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

security: preserve privacyIDEA key custody and restore the actual PVC
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a1346b-72a4-7d40-b8bd-8b8ca346cfa2
This commit is contained in:
tegwick 2026-10-13 21:46:41 +02:00
parent 16021d3953
commit 1cbe00fd78
15 changed files with 937 additions and 182 deletions
Download patch file Download diff file Expand all files Collapse all files

2
sso-mfa/WORKPLAN.md
View file

@ -52,7 +52,7 @@ Hostnames: kc.coulomb.social (KeyCape), auth.coulomb.social (Authelia), lldap.co
- [ ] `./create-secrets.sh` — create privacyidea-config Secret in mfa namespace (may already exist; check first)
- [ ] `kubectl apply -f pvc.yaml configmap.yaml middleware.yaml deployment.yaml ingress.yaml`
- [ ] Wait for pod Running/Ready (up to 3 min — DB migrations run on first boot)
- [ ] `./enckey-bootstrap.sh` — extract enckey+auditkeys, store in KeePassXC, create DR Secrets
- [ ] `bash ./enckey-bootstrap.sh /absolute/private/custody` — extract enckey+auditkeys, store in KeePassXC, create DR Secrets
- [ ] `./bootstrap-admin.sh` — create/verify pi-admin (current KeyCape profile)
- [ ] Log in to pink.coulomb.social, enroll TOTP for pi-admin, verify MFA challenge
- [ ] Run `../verify-t04.sh` — 0 FAILs

3
sso-mfa/bootstrap/apply-service-secrets.py
View file

@ -176,7 +176,8 @@ def service_values(component, values):
def publish(namespace, name, desired, old, allowed):
encoded = {k: base64.b64encode(v.encode()).decode() for k, v in desired.items()}
encoded = {k: base64.b64encode(v.encode() if isinstance(v, str) else v).decode()
for k, v in desired.items()}
if old is None:
if allowed: raise Refused() # Rotation may not silently create missing state.
obj = {'apiVersion': 'v1', 'kind': 'Secret', 'type': 'Opaque',

4
sso-mfa/bootstrap/creds-apply.sh
View file

@ -108,8 +108,8 @@ echo "────────────────────────
echo "SKIPPED: keycape (requires RPF-WP-0040 scoped factor delivery)"
echo ""
echo " After privacyIDEA is Running and bootstrapped:"
echo " 1. TIME-SENSITIVE: run enckey-bootstrap.sh while pod is live:"
echo " bash $K8S_DIR/privacyidea/enckey-bootstrap.sh"
echo " 1. Export existing keys from the verified Ready privacyIDEA pod:"
echo " bash $K8S_DIR/privacyidea/enckey-bootstrap.sh $SECRETS_DIR"
echo ""
echo " 2. Create the pi-admin user:"
echo " bash $K8S_DIR/privacyidea/bootstrap-admin.sh"

6
sso-mfa/bootstrap/creds-bootstrap-agent.sh
View file

@ -2,7 +2,7 @@
# creds-bootstrap-agent.sh — attended bootstrap with public-recipient encryption
#
# Usage:
# bash sso-mfa/bootstrap/creds-bootstrap-agent.sh [--dry-run] [--resume]
# bash sso-mfa/bootstrap/creds-bootstrap-agent.sh [--dry-run] [--resume] [--initialize-audit-keys]
# make creds-agent-init
#
# Runs in the operator terminal and ends at the encrypted recovery handoff
@ -29,9 +29,11 @@ STATE_FILE="$SCRIPT_DIR/creds-state.yaml"
AGE_KEY="${NETKINGDOM_CUSTODY_AGE_KEY_FILE:-}"
DRY_RUN=false
AUDIT_ARGS=()
for arg in "$@"; do
case "$arg" in
--dry-run) DRY_RUN=true ;;
--initialize-audit-keys) AUDIT_ARGS=(--initialize-audit-keys) ;;
--resume) ;; # Phase state already controls resume.
*) printf '%s\n' 'ERROR: unsupported bootstrap argument.' >&2; exit 2 ;;
esac
@ -264,7 +266,7 @@ else
# Run enckey bootstrap
if [[ "$(state_get enckey_bootstrapped)" != "true" ]]; then
log "running enckey-bootstrap.sh..."
(cd "$K8S_DIR/privacyidea" && bash enckey-bootstrap.sh "$SECRETS_DIR")
(cd "$K8S_DIR/privacyidea" && bash enckey-bootstrap.sh "$SECRETS_DIR" "${AUDIT_ARGS[@]}")
state_set "enckey_bootstrapped" "true"
ok "enckey bootstrapped"
else

6
sso-mfa/bootstrap/gen-secrets.sh
View file

@ -13,7 +13,7 @@
#
# PI_ENCFILE is NOT generated here — it must be produced inside the privacyIDEA
# container after deployment:
# kubectl exec -n mfa <pi-pod> -- pi-manage create_enckey
# Follow ../k8s/privacyidea/KEY-RECOVERY.md; do not generate missing recovery keys.
# Extract the resulting file, store it in KeePassXC as a binary attachment on the
# privacyIDEA/PI_ENCFILE entry, then create the k8s secret from it.
@ -60,8 +60,8 @@ PI_DB_PASSWORD=$PI_DB_PASS
PI_ADMIN_PASSWORD=$PI_ADMIN_PASS
# PI_ENCFILE: generate with:
# kubectl exec -n mfa <pi-pod> -- pi-manage create_enckey
# kubectl cp -n mfa <pi-pod>:/etc/privacyidea/enckey ./secrets/privacyidea/pi.enc
# Follow ../k8s/privacyidea/KEY-RECOVERY.md; do not generate missing recovery keys.
# Export existing keys with enckey-bootstrap.sh and an explicit private custody root.
# Then store pi.enc as a binary attachment in KeePassXC → net-kingdom/privacyIDEA/PI_ENCFILE
EOF

29
sso-mfa/k8s/backup/DR-RUNBOOK.md
View file

@ -105,24 +105,15 @@ kubectl scale deployment/authelia -n sso --replicas=1
kubectl rollout status deployment/authelia -n sso --timeout=120s
```
### privacyIDEA enckey
### privacyIDEA encryption and audit keys
```bash
# If the enckey is lost, restore it from KeePassXC binary attachment PI_ENCFILE.
# Extract it to a local file first, then:
kubectl create secret generic privacyidea-enckey \
--from-file=PI_ENCFILE=./pi.enc \
--namespace mfa \
--dry-run=client -o yaml | kubectl apply -f -
# Restart privacyIDEA to pick up the restored key
kubectl rollout restart deployment/privacyidea -n mfa
# If the enckey is truly lost and unrecoverable:
# All enrolled MFA tokens are invalid.
# Generate a new enckey with: kubectl exec -n mfa ... -- pi-manage create_enckey
# All users must re-enroll their TOTP/hardware tokens.
```
Use the [privacyIDEA key custody/PVC recovery procedure](../privacyidea/KEY-RECOVERY.md).
The authoritative paths are on `privacyidea-data`, not a mounted Secret. Restore
all three files from the matching independent backup into that PVC with the app
stopped, then perform native database/token/audit acceptance. The DR Secret field
is `enckey`; `PI_ENCFILE` names the application config setting and historic safe
entry. The previous Secret-only restart instructions are withdrawn. Never generate
a fresh encryption key as an automatic response to a missing recovery key.
---
@ -169,7 +160,9 @@ kubectl rollout status deployment/authelia -n sso --timeout=120s
kubectl rollout status deployment/keycape -n sso --timeout=60s
# 6. Re-run bootstrap scripts if PVC data was lost
cd privacyidea && ./enckey-bootstrap.sh && ./bootstrap-admin.sh && ./bootstrap-realm.sh
# Restore the original key set first using privacyidea/KEY-RECOVERY.md.
# Only then follow the reviewed account/realm reconciliation procedure.
cd privacyidea && ./enckey-bootstrap.sh /absolute/private/custody
cd ../lldap && ./bootstrap-users.sh && ./break-glass.sh
# Verify owner-managed factor renewal, ESO delivery and mounted reload (RPF-WP-0040).
# Do not mint a personal admin token or replace existing KeyCape configuration.

144
sso-mfa/k8s/privacyidea/KEY-RECOVERY.md Normal file
View file

@ -0,0 +1,144 @@
# privacyIDEA key custody and PVC recovery
This is the source procedure for the admitted `mfa/privacyidea` profile:
`privacyidea-data` mounted at `/etc/privacyidea`, `fsGroup: 999`, and the key paths
in `configmap.yaml`. It does not authorize live execution or establish successful
recovery. Use an exact owner-approved ceremony on the verified cluster, with an
exclusive maintenance window, independent database/key backups and current
consumer acceptance. On Railiance, use the approved node execution path; do not
infer the target from an ambient workstation kube context.
## Export an existing key set
Create an owned private custody root and its `privacyidea` child (both mode 0700)
outside Git for an independent export. The bootstrap caller can also use its
existing ignored private service-file directory. Run:
```bash
bash sso-mfa/k8s/privacyidea/enckey-bootstrap.sh /absolute/private/custody
```
The wrapper verifies one Ready deployment-owned pod using the expected Bound
PVC. It reads all three files through a captured, bounded transport, validates the
96-byte encryption key and matching RSA audit pair, and rechecks pod/PVC identity.
It never generates a missing encryption key. Exported files are private mode
0600, with atomic no-overwrite publication; identical existing files are verified
and differing, public, linked or nonregular files are refused.
| Local custody file | Runtime PVC path | DR Secret and field |
| --- | --- | --- |
| `privacyidea/pi.enc` | `/etc/privacyidea/enckey` | `privacyidea-enckey`: `enckey` |
| `privacyidea/private.pem` | `/etc/privacyidea/private.pem` | `privacyidea-auditkeys`: `private.pem` |
| `privacyidea/public.pem` | `/etc/privacyidea/public.pem` | `privacyidea-auditkeys`: `public.pem` |
`PI_ENCFILE` is the application configuration setting and the historic password
safe entry name, **not** the Secret data field. A legacy differing Secret must be
reconciled under its owner's migration procedure; the wrapper does not rename,
overwrite or rotate it automatically.
DR Secrets are created or verified against matching data. Existing extra fields
and metadata are retained. Removal of a last-applied annotation uses UID,
resource-version and data preconditions; no client-side apply or conflict retry.
Controller-managed or immutable Secrets are refused. Local copies remain when an
API operation fails. The two Secret writes and three file publications are not a
transaction: failure may leave a partial result or an unverified remote write.
Inspect and reconcile before retrying. Do not delete the retained copies to make
a retry pass.
Store all three files together in independently held encrypted/off-host custody,
linked to the correct database backup and instance. Verify that backup before
removing successful working copies. Kubernetes DR Secrets in the same cluster
are not an independent recovery backup. Historical exposure/revocation remains
separate incident work.
## First-bootstrap audit pair
Only for a reviewed, genuinely new instance, when its encryption key already
exists and both audit files and all local/cluster DR copies are absent:
```bash
bash sso-mfa/k8s/privacyidea/enckey-bootstrap.sh /absolute/private/custody --initialize-audit-keys
```
This explicitly creates only the absent RSA audit pair using no-overwrite
publication, then exports it. The encryption key is never generated or replaced.
An existing single audit file is a partial/recovery state and is refused. A
failed pair publication may retain one file; reconcile it rather than generate
another pair. The flag is an explicit action selection, not proof of first-use
identity or permission. It must not be used to repair lost historical audit keys.
`creds-bootstrap-agent.sh` accepts the same flag explicitly; normal bootstrap
export does not infer permission to initialize keys.
## Restore into the actual PVC
1. Select the approved database backup and its matching independent key set.
Decrypt/retrieve them through approved custody into the private directory
above. Validate provenance independently; cryptographic format checks cannot
establish that a key belongs to a database. Do not substitute fresh keys.
2. Under owner control, stop privacyIDEA and ensure its Deployment has desired and
observed replicas zero. Wait until **all** pods mounting `privacyidea-data`
have disappeared, including terminating pods. Pause any reconciler that would
restart it under its own approved maintenance procedure. Do not start the app
on an empty/restored PVC before the original keys are present.
3. Create/bind the intended replacement `privacyidea-data` PVC if needed. Retain
its exact UID and an independently reviewed immutable digest of the admitted
privacyIDEA image containing Python and cryptography. Render the helper pod:
```bash
python3 sso-mfa/k8s/privacyidea/key-custody.py render-pod \
--pvc-uid APPROVED_PVC_UID \
--image ghcr.io/gpappsoft/privacyidea-docker@sha256:APPROVED_IMAGE_DIGEST \
> /private/recovery-pod.json
```
Replace both placeholders with reviewed values; the renderer rejects them
literally. Review/dry-run the resulting non-secret manifest, then create it
through the approved cluster execution path. It mounts the **actual data PVC**,
sleeps instead of starting privacyIDEA, has no service-account token mount,
and runs for at most one hour. It does not consume either DR Secret. The helper
requires root/limited file-ownership capabilities to restore root:999 files;
this is an attended maintenance pod, not a new permanent service.
4. Wait for that exact `privacyidea-key-recovery` pod to be Ready, then run:
```bash
bash sso-mfa/k8s/privacyidea/enckey-restore.sh /absolute/private/custody \
--pvc-uid APPROVED_PVC_UID
```
The wrapper checks the stopped Deployment, PVC UID, sole mounting pod, fixed
sleep command, immutable image, container and mount before transferring keys
on captured stdin. It validates the full local key set before any cluster
command. It checks every existing remote file before writing; differences
refuse the operation without overwriting them. Missing files are published
without replacement, read back, and pod/PVC identity is checked again.
New private files use root:999 mode 0640 so the declared service fsGroup can
read them; the public file uses mode 0644. Existing matching files retain their
modes/ownership. Unknown remote outcomes are failures, never automatic retries.
5. Confirm the receipt and state, then delete only the exact helper pod created
in this ceremony, with its recorded UID/preconditions. Wait for its removal
before restoring the approved application replica count/reconciler state.
Perform native database decryption, existing-token authentication and audit
signing/verification against the recovered instance. Retain safe evidence and
independent recovery material; remove temporary working plaintext afterward.
The helper does not create/stop/delete workloads or restart the application.
These lifecycle actions stay explicit in the reviewed maintenance plan. A Secret
creation followed by rollout restart cannot restore these PVC files. The former
unmounted BusyBox helper example is withdrawn.
## Verification and limits
Tests use disposable local files/keys and a synthetic API, including an actual
privacyIDEA 3.12.2 security-module decryption round trip after restore. Its installed
`privacyidea/cli/pimanage/pi_setup.py` establishes the 96-byte key and RSA PEM
formats; [upstream pi-manage documentation](https://privacyidea.readthedocs.io/en/latest/installation/system/pi-manage.html)
describes the setup operations. Package verification is not deployed-image proof.
Cooperative directory locks and no-overwrite links are not protection against
hostile same-UID/container writers. Pod/PVC checks before and after exec are not
an atomic Kubernetes UID-bound exec. Use the exclusive owner-controlled window;
changes detected after transfer produce an unverified outcome. A crash/SIGKILL
can leave staging or partial files. File permissions and terminal checks provide
neither task authority nor full agent credential isolation. Native off-host
recovery, image admission, backup lineage and current-consumer acceptance remain
required by NK-WP-0049-T03.

44
sso-mfa/k8s/privacyidea/README.md
View file

@ -63,20 +63,19 @@ Common causes:
### Step 3 — Extract key material and create DR Secrets
Run **once** after the pod reaches `Running`:
Export the existing key set from the verified Ready pod into an owned private
custody root containing a private `privacyidea` subdirectory:
```bash
./enckey-bootstrap.sh
bash enckey-bootstrap.sh /absolute/private/custody
```
This generates RSA audit keys (if not already created), extracts the encryption
key and audit keys from the pod, and creates two K8s Secrets as disaster-recovery
copies:
- `privacyidea-enckey`
- `privacyidea-auditkeys`
Follow the printed instructions to store the key files in KeePassXC, then shred
the local copies.
This preserves existing local copies and creates or verifies matching DR Secrets:
`privacyidea-enckey` (`enckey`) and `privacyidea-auditkeys` (`private.pem`,
`public.pem`). Missing keys are refused. Only an explicitly reviewed first
bootstrap can request `--initialize-audit-keys`; it never generates an encryption
key. Follow [key custody and PVC recovery](KEY-RECOVERY.md) for private export,
independent backup, failure handling and recovery acceptance.
---
@ -201,21 +200,10 @@ Config → Policies → New policy:
## Disaster recovery
If the `privacyidea-data` PVC is lost:
1. Create a new PVC with `pvc.yaml`.
2. Restore enckey and audit keys from KeePassXC:
```bash
# Copy pi.enc and private/public.pem from KeePassXC into a temporary pod
kubectl run restore-helper --image=busybox -n mfa --restart=Never -- sleep 3600
kubectl cp ./pi.enc mfa/restore-helper:/etc/privacyidea/enckey
kubectl cp ./private.pem mfa/restore-helper:/etc/privacyidea/private.pem
kubectl cp ./public.pem mfa/restore-helper:/etc/privacyidea/public.pem
kubectl delete pod -n mfa restore-helper
```
Or, if the K8s Secrets survived (created by `enckey-bootstrap.sh`):
```bash
kubectl get secret privacyidea-enckey -n mfa -o jsonpath='{.data.enckey}' | base64 -d > pi.enc
```
3. Restart the privacyIDEA deployment — it will run DB migrations and use the
restored key material.
Follow [key custody and PVC recovery](KEY-RECOVERY.md). Restore the original
matching encryption/audit files into the actual `privacyidea-data` PVC while the
application is stopped, using the reviewed mounted recovery pod. Creating a
Secret and restarting cannot restore these files; the deployment does not mount
the DR Secrets. The former unmounted helper-pod example is withdrawn. Native
recovery acceptance requires the matching database backup, existing-token checks
and audit-signature verification before closure.

122
sso-mfa/k8s/privacyidea/enckey-bootstrap.sh
View file

@ -1,121 +1,5 @@
#!/usr/bin/env bash
# enckey-bootstrap.sh — extract privacyIDEA key material and create DR Secrets
#
# Run ONCE after the first successful pod start (deployment.yaml applied and Running).
#
# What it does:
# 1. Finds the running privacyIDEA pod.
# 2. Generates RSA audit keys inside the pod if they do not exist yet.
# 3. Extracts: enckey, private.pem, public.pem → local files.
# 4. Creates K8s Secrets privacyidea-enckey and privacyidea-auditkeys.
# These are disaster-recovery copies — the authoritative key material
# lives in the privacyidea-data PVC.
# 5. Prints instructions to copy the extracted files into KeePassXC.
#
# Usage:
# ./enckey-bootstrap.sh [secrets-dir]
#
# <secrets-dir> is where local copies are written (default: ../../bootstrap/secrets).
# Files in secrets-dir are sensitive — shred them after entering into KeePassXC.
#
# Re-running is safe: kubectl apply is idempotent; pi-manage create_audit_keys
# does not overwrite existing keys.
# Explicit private export; first-bootstrap audit creation requires the named flag.
set +x
set -euo pipefail
NAMESPACE="mfa"
SECRETS_DIR="${1:-../../bootstrap/secrets}"
PI_DIR="$SECRETS_DIR/privacyidea"
mkdir -p "$PI_DIR"
# ── 1. Find the running pod ───────────────────────────────────────────────────
echo "Looking for running privacyIDEA pod in namespace: $NAMESPACE"
PI_POD=$(kubectl get pod -n "$NAMESPACE" \
-l app.kubernetes.io/name=privacyidea \
--field-selector=status.phase=Running \
-o jsonpath='{.items[0].metadata.name}' 2>/dev/null || echo "")
if [[ -z "$PI_POD" ]]; then
echo "ERROR: no Running privacyIDEA pod found in namespace '$NAMESPACE'." >&2
echo " Apply the manifests first and wait for the pod to reach Running state:" >&2
echo " kubectl get pods -n $NAMESPACE -w" >&2
exit 1
fi
echo "Found pod: $PI_POD"
# ── 2. Generate audit keys if missing ────────────────────────────────────────
echo ""
echo "Checking for audit keys..."
if kubectl exec -n "$NAMESPACE" "$PI_POD" -- test -f /etc/privacyidea/private.pem 2>/dev/null; then
echo " Audit keys already exist — skipping generation."
else
echo " Generating audit keys (pi-manage create_audit_keys)..."
kubectl exec -n "$NAMESPACE" "$PI_POD" -- pi-manage create_audit_keys
echo " Done."
fi
# ── 3. Check enckey exists (auto-generated by PI on first DB init) ────────────
echo ""
echo "Checking for encryption key..."
if ! kubectl exec -n "$NAMESPACE" "$PI_POD" -- test -f /etc/privacyidea/enckey 2>/dev/null; then
echo " enckey not found — generating (pi-manage create_enckey)..."
kubectl exec -n "$NAMESPACE" "$PI_POD" -- pi-manage create_enckey
echo " Done."
else
echo " enckey exists."
fi
# ── 4. Extract key material to local files ────────────────────────────────────
echo ""
echo "Extracting key material..."
kubectl exec -n "$NAMESPACE" "$PI_POD" -- \
cat /etc/privacyidea/enckey > "$PI_DIR/pi.enc"
echo " Wrote: $PI_DIR/pi.enc"
kubectl exec -n "$NAMESPACE" "$PI_POD" -- \
cat /etc/privacyidea/private.pem > "$PI_DIR/private.pem"
echo " Wrote: $PI_DIR/private.pem"
kubectl exec -n "$NAMESPACE" "$PI_POD" -- \
cat /etc/privacyidea/public.pem > "$PI_DIR/public.pem"
echo " Wrote: $PI_DIR/public.pem"
# ── 5. Create K8s Secrets (DR copies) ────────────────────────────────────────
echo ""
echo "Creating K8s Secret: privacyidea-enckey (namespace: $NAMESPACE)"
kubectl create secret generic privacyidea-enckey \
--namespace="$NAMESPACE" \
--from-file=enckey="$PI_DIR/pi.enc" \
--dry-run=client -o yaml | kubectl apply -f -
echo "Creating K8s Secret: privacyidea-auditkeys (namespace: $NAMESPACE)"
kubectl create secret generic privacyidea-auditkeys \
--namespace="$NAMESPACE" \
--from-file=private.pem="$PI_DIR/private.pem" \
--from-file=public.pem="$PI_DIR/public.pem" \
--dry-run=client -o yaml | kubectl apply -f -
# ── 6. Summary and KeePassXC instructions ─────────────────────────────────────
echo ""
echo "════════════════════════════════════════════════════════════"
echo " Key material extracted and K8s Secrets created."
echo "════════════════════════════════════════════════════════════"
echo ""
echo "IMPORTANT — Store the extracted files in KeePassXC NOW:"
echo ""
echo " KeePassXC group: net-kingdom/privacyIDEA"
echo " Entry: PI_ENCFILE"
echo " → Attach $PI_DIR/pi.enc as a binary attachment."
echo " Entry: Audit Keys"
echo " → Attach $PI_DIR/private.pem and $PI_DIR/public.pem"
echo ""
echo " Then shred the local copies:"
echo " find '$PI_DIR' -name '*.enc' -o -name '*.pem' | xargs shred -u"
echo ""
echo "DR restore: if the privacyidea-data PVC is lost, copy these files"
echo " back into a fresh PVC before starting the pod."
echo ""
echo "Next step: run ./bootstrap-admin.sh"
exec python3 "$(dirname -- "${BASH_SOURCE[0]}")/key-custody.py" export "$@"

5
sso-mfa/k8s/privacyidea/enckey-restore.sh Normal file
View file

@ -0,0 +1,5 @@
#!/usr/bin/env bash
# Restore private recovery files into the reviewed helper pod's actual data PVC.
set +x
set -euo pipefail
exec python3 "$(dirname -- "${BASH_SOURCE[0]}")/key-custody.py" restore "$@"

138
sso-mfa/k8s/privacyidea/key-custody-remote.py Normal file
View file

@ -0,0 +1,138 @@
#!/usr/bin/env python3
"""In-pod key transport. Only invoke through key-custody.py (captured output)."""
import base64
import fcntl
import json
import os
from pathlib import Path
import secrets
import stat
import sys
FILES = ('enckey', 'private.pem', 'public.pem')
LIMIT = 65536
class Refused(Exception):
pass
def validate(values):
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
if set(values) != set(FILES) or len(values['enckey']) != 96:
raise Refused()
if any(not value or len(value) > LIMIT for value in values.values()):
raise Refused()
private = serialization.load_pem_private_key(values['private.pem'], password=None)
public = serialization.load_pem_public_key(values['public.pem'])
if (not isinstance(private, rsa.RSAPrivateKey) or private.key_size < 2048
or not isinstance(public, rsa.RSAPublicKey)
or private.public_key().public_numbers() != public.public_numbers()):
raise Refused()
def read(directory, name):
fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
try:
info = os.fstat(fd)
forbidden = 0o022 if name == 'public.pem' else 0o027
if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_mode & forbidden:
raise Refused()
with os.fdopen(os.dup(fd), 'rb') as stream:
data = stream.read(LIMIT + 1)
after = os.stat(name, dir_fd=directory, follow_symlinks=False)
if len(data) > LIMIT or any(getattr(info, k) != getattr(after, k)
for k in ('st_dev', 'st_ino', 'st_size', 'st_mtime_ns', 'st_ctime_ns')):
raise Refused()
return data
finally:
os.close(fd)
def install(directory, values, *, uid=0, gid=999):
"""Verify every existing file before creating anything; never replace."""
missing = []
for name, value in values.items():
try:
if read(directory, name) != value:
raise Refused()
except FileNotFoundError:
missing.append(name)
for name in missing:
temp = '.key-custody-' + secrets.token_hex(12)
fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
0o600, dir_fd=directory)
try:
with os.fdopen(fd, 'wb') as stream:
stream.write(values[name]); stream.flush()
os.fchown(stream.fileno(), uid, gid)
os.fchmod(stream.fileno(), 0o644 if name == 'public.pem' else 0o640)
os.fsync(stream.fileno())
os.link(temp, name, src_dir_fd=directory, dst_dir_fd=directory, follow_symlinks=False)
os.unlink(temp, dir_fd=directory)
os.fsync(directory)
finally:
try:
os.unlink(temp, dir_fd=directory)
except FileNotFoundError:
pass
if any(read(directory, name) != value for name, value in values.items()):
raise Refused()
def operate(request, root=Path('/etc/privacyidea'), *, uid=0, gid=999):
if any(p.is_symlink() for p in (root, *root.parents)):
raise Refused()
directory = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
try:
# The admitted service uses fsGroup 999, which may make its PVC group-writable.
if os.fstat(directory).st_mode & 0o002:
raise Refused()
fcntl.flock(directory, fcntl.LOCK_EX | fcntl.LOCK_NB)
if request == {'action': 'initialize-audit'}:
# This explicit first-bootstrap action never creates/replaces enckey.
enckey = read(directory, 'enckey')
if len(enckey) != 96:
raise Refused()
for name in ('private.pem', 'public.pem'):
try:
os.stat(name, dir_fd=directory, follow_symlinks=False)
except FileNotFoundError:
continue
raise Refused()
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
values = {'enckey': enckey,
'private.pem': private.private_bytes(serialization.Encoding.PEM,
serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption()),
'public.pem': private.public_key().public_bytes(serialization.Encoding.PEM,
serialization.PublicFormat.SubjectPublicKeyInfo)}
validate(values)
install(directory, values, uid=uid, gid=gid)
return {'result': 'initialized'}
if request == {'action': 'export'}:
values = {name: read(directory, name) for name in FILES}
validate(values)
return {name: base64.b64encode(value).decode() for name, value in values.items()}
if set(request) == {'action', 'keys'} and request['action'] == 'restore':
values = {name: base64.b64decode(value, validate=True) for name, value in request['keys'].items()}
validate(values)
install(directory, values, uid=uid, gid=gid)
return {'result': 'restored'}
raise Refused()
finally:
os.close(directory)
if __name__ == '__main__':
try:
raw = sys.stdin.buffer.read(4 * LIMIT + 1)
if len(raw) > 4 * LIMIT:
raise Refused()
print(json.dumps(operate(json.loads(raw))))
except Exception:
# Transport may carry key bytes; diagnostics must never serialize them.
print('Key custody operation refused or failed.', file=sys.stderr)
sys.exit(1)

254
sso-mfa/k8s/privacyidea/key-custody.py Normal file
View file

@ -0,0 +1,254 @@
#!/usr/bin/env python3
"""Captured key export/restore for the admitted privacyIDEA PVC profile."""
import argparse
import base64
import importlib.util
import json
import os
from pathlib import Path
import re
import secrets
import sys
HERE = Path(__file__).resolve().parent
def load(name, path):
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
common = load('service_custody', HERE.parents[1] / 'bootstrap/apply-service-secrets.py')
remote = load('key_custody_remote', HERE / 'key-custody-remote.py')
Refused = common.Refused
LOCAL = {'enckey': 'pi.enc', 'private.pem': 'private.pem', 'public.pem': 'public.pem'}
TARGETS = {'privacyidea-enckey': ('enckey',), 'privacyidea-auditkeys': ('private.pem', 'public.pem')}
RECOVERY_POD = 'privacyidea-key-recovery'
SLEEP = ['python3', '-c', 'import time; time.sleep(3600)']
IMAGE = re.compile(r'ghcr\.io/gpappsoft/privacyidea-docker@sha256:[a-f0-9]{64}\Z')
UID = re.compile(r'[a-zA-Z0-9-]{1,128}\Z')
def get(kind, name=None, missing=False):
args = ['kubectl', 'get', kind]
if name:
args.append(name)
args += ['-n', 'mfa', '-o', 'json']
if missing:
args.append('--ignore-not-found')
raw = common.run(args)
return json.loads(raw) if raw.strip() else None
def meta(obj, kind, name=None):
if not isinstance(obj, dict) or obj.get('kind') != kind:
raise Refused()
m = obj.get('metadata', {})
if (m.get('namespace') != 'mfa' or (name and m.get('name') != name)
or not UID.fullmatch(str(m.get('uid', ''))) or m.get('deletionTimestamp')):
raise Refused()
return m
def mounts_data(spec, container):
volumes = {v['name']: v for v in spec.get('volumes', [])}
return any(v.get('mountPath') == '/etc/privacyidea' and not v.get('subPath')
and not v.get('readOnly') and volumes.get(v['name'], {}).get('persistentVolumeClaim', {}).get('claimName') == 'privacyidea-data'
for v in container.get('volumeMounts', []))
def pod_uses_data(pod):
return any(v.get('persistentVolumeClaim', {}).get('claimName') == 'privacyidea-data'
for v in pod.get('spec', {}).get('volumes', []))
def context(action, expected_uid=None):
deployment = get('deployment', 'privacyidea')
dm = meta(deployment, 'Deployment', 'privacyidea')
template = deployment['spec']['template']['spec']
container = next(c for c in template['containers'] if c['name'] == 'privacyidea')
if template.get('securityContext', {}).get('fsGroup') != 999 or not mounts_data(template, container):
raise Refused()
pvc = get('pvc', 'privacyidea-data')
pm = meta(pvc, 'PersistentVolumeClaim', 'privacyidea-data')
if pvc.get('status', {}).get('phase') != 'Bound' or (expected_uid and pm['uid'] != expected_uid):
raise Refused()
listing = get('pods')
if listing.get('kind') not in ('PodList', 'List'):
raise Refused()
users = [pod for pod in listing['items'] if pod_uses_data(pod)]
if len(users) != 1:
raise Refused()
pod = users[0]; podmeta = meta(pod, 'Pod')
spec = pod['spec']
if pod.get('status', {}).get('phase') != 'Running':
raise Refused()
if action == 'restore':
if (deployment['spec'].get('replicas', 1) != 0
or any(deployment.get('status', {}).get(k, 0) for k in ('replicas', 'readyReplicas', 'availableReplicas', 'updatedReplicas'))
or deployment.get('status', {}).get('observedGeneration', 0) < dm.get('generation', 1)):
raise Refused()
if (podmeta['name'] != RECOVERY_POD or podmeta.get('annotations', {}).get('net-kingdom/recovery-pvc-uid') != expected_uid
or spec.get('automountServiceAccountToken') is not False
or any(spec.get(k) for k in ('hostNetwork', 'hostPID', 'hostIPC', 'initContainers', 'ephemeralContainers'))
or len(spec.get('containers', [])) != 1 or len(spec.get('volumes', [])) != 1):
raise Refused()
c = spec['containers'][0]
if (c.get('name') != 'recovery' or not IMAGE.fullmatch(c.get('image', ''))
or c.get('command') != SLEEP or c.get('args') or c.get('env') or c.get('envFrom')
or c.get('securityContext', {}).get('privileged')
or not mounts_data(spec, c)):
raise Refused()
else:
c = next(c for c in spec['containers'] if c['name'] == 'privacyidea')
if not mounts_data(spec, c):
raise Refused()
owners = podmeta.get('ownerReferences', [])
if len(owners) != 1 or owners[0].get('kind') != 'ReplicaSet' or not owners[0].get('controller'):
raise Refused()
rs = get('replicaset', owners[0]['name']); rm = meta(rs, 'ReplicaSet', owners[0]['name'])
if rm['uid'] != owners[0]['uid'] or not any(o.get('uid') == dm['uid'] and o.get('kind') == 'Deployment' and o.get('controller') for o in rm.get('ownerReferences', [])):
raise Refused()
if not any(s.get('name') == c['name'] and s.get('ready') for s in pod['status'].get('containerStatuses', [])):
raise Refused()
return {'pod': podmeta['name'], 'pod_uid': podmeta['uid'], 'pvc_uid': pm['uid'],
'deployment_uid': dm['uid'], 'container': c['name']}
def invoke(ctx, request):
source = (HERE / 'key-custody-remote.py').read_text()
raw = common.run(['kubectl', 'exec', '-i', '-n', 'mfa', ctx['pod'], '-c', ctx['container'],
'--', 'python3', '-c', source], json.dumps(request).encode())
return json.loads(raw)
def read_backups():
result = {}
for name in TARGETS:
obj = get('secret', name, missing=True)
result[name] = common.decode_secret(json.dumps(obj), 'mfa', name) if obj is not None else None
return result
def verify_backups(backups, values):
for name, keys in TARGETS.items():
old = backups[name]
if old is not None and any(old['data'].get(k) != base64.b64encode(values[k]).decode() for k in keys):
raise Refused()
def local_values(directory):
result = {name: common.read_private(directory, local) for name, local in LOCAL.items()}
remote.validate(result)
return result
def save_local(directory, values):
# All differences are checked before creating any file. Existing copies survive.
missing = []
for name, local in LOCAL.items():
try:
if common.read_private(directory, local) != values[name]:
raise Refused()
except FileNotFoundError:
missing.append((name, local))
for name, local in missing:
temporary = '.key-export-' + secrets.token_hex(12)
fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory)
try:
with os.fdopen(fd, 'wb') as stream:
stream.write(values[name]); stream.flush(); os.fsync(stream.fileno())
os.link(temporary, local, src_dir_fd=directory, dst_dir_fd=directory, follow_symlinks=False)
os.unlink(temporary, dir_fd=directory); os.fsync(directory)
finally:
try:
os.unlink(temporary, dir_fd=directory)
except FileNotFoundError:
pass
if local_values(directory) != values:
raise Refused()
def perform(action, root, expected_uid=None, initialize=False):
if action == 'restore' and (not expected_uid or not UID.fullmatch(expected_uid)):
raise Refused()
with common.custody(root, ('privacyidea',)) as directories:
directory = directories['privacyidea']
values = local_values(directory) if action == 'restore' else None
ctx = context(action, expected_uid)
if action == 'restore':
result = invoke(ctx, {'action': 'restore', 'keys': {k: base64.b64encode(v).decode() for k, v in values.items()}})
if result != {'result': 'restored'} or context(action, expected_uid) != ctx:
raise Refused()
else:
backups = read_backups()
if initialize:
if any(value is not None for value in backups.values()):
raise Refused()
# Independent local recovery copies also prohibit first-bootstrap creation.
for name in LOCAL.values():
try:
os.stat(name, dir_fd=directory, follow_symlinks=False)
except FileNotFoundError:
continue
raise Refused()
if invoke(ctx, {'action': 'initialize-audit'}) != {'result': 'initialized'}:
raise Refused()
encoded = invoke(ctx, {'action': 'export'})
values = {k: base64.b64decode(v, validate=True) for k, v in encoded.items()}
remote.validate(values)
if context(action) != ctx:
raise Refused()
verify_backups(backups, values)
save_local(directory, values)
for name, keys in TARGETS.items():
common.publish('mfa', name, {k: values[k] for k in keys}, backups[name], set())
print('Key custody operation verified. Retain independent recovery copies; native service acceptance remains required.')
def render(image, pvc_uid):
if not image or not IMAGE.fullmatch(image) or not pvc_uid or not UID.fullmatch(pvc_uid):
raise Refused()
return {'apiVersion': 'v1', 'kind': 'Pod',
'metadata': {'name': RECOVERY_POD, 'namespace': 'mfa',
'annotations': {'net-kingdom/recovery-pvc-uid': pvc_uid}},
'spec': {'restartPolicy': 'Never', 'automountServiceAccountToken': False,
'securityContext': {'runAsUser': 0, 'runAsGroup': 999, 'fsGroup': 999},
'containers': [{'name': 'recovery', 'image': image, 'command': SLEEP,
'securityContext': {'allowPrivilegeEscalation': False, 'readOnlyRootFilesystem': True,
'capabilities': {'drop': ['ALL'], 'add': ['CHOWN', 'DAC_OVERRIDE', 'FOWNER']}},
'resources': {'requests': {'cpu': '10m', 'memory': '64Mi'}, 'limits': {'cpu': '100m', 'memory': '128Mi'}},
'volumeMounts': [{'name': 'data', 'mountPath': '/etc/privacyidea'}]}],
'volumes': [{'name': 'data', 'persistentVolumeClaim': {'claimName': 'privacyidea-data'}}]}}
def main():
class QuietParser(argparse.ArgumentParser):
def error(self, message):
raise Refused()
parser = QuietParser(description=__doc__)
parser.add_argument('action', choices=['export', 'restore', 'render-pod'])
parser.add_argument('root', nargs='?')
parser.add_argument('--pvc-uid')
parser.add_argument('--image')
parser.add_argument('--initialize-audit-keys', action='store_true')
args = parser.parse_args()
if args.action == 'render-pod':
if args.root or args.initialize_audit_keys:
raise Refused()
print(json.dumps(render(args.image, args.pvc_uid), indent=2))
else:
if not args.root or args.image or (args.action == 'restore' and args.initialize_audit_keys) or (args.action == 'export' and args.pvc_uid):
raise Refused()
perform(args.action, args.root, args.pvc_uid, args.initialize_audit_keys)
if __name__ == '__main__':
try:
main()
except Exception:
print('ERROR: key custody refused or failed; diagnostics suppressed. A failed write may have partial effects. '
'Retain copies and reconcile target state before retrying.', file=sys.stderr)
sys.exit(1)

6
sso-mfa/k8s/verify-t04.sh
View file

@ -147,7 +147,7 @@ for secret in privacyidea-enckey privacyidea-auditkeys; do
if kubectl get secret "$secret" -n "$NAMESPACE" &>/dev/null; then
pass "Secret $secret exists"
else
warn "Secret $secret not found — run enckey-bootstrap.sh"
warn "Secret $secret not found — use privacyidea/KEY-RECOVERY.md for private export"
fi
done
@ -173,13 +173,13 @@ if [[ -n "$PI_POD" ]]; then
if kubectl exec -n "$NAMESPACE" "$PI_POD" -- test -f /etc/privacyidea/enckey 2>/dev/null; then
pass "enckey present in pod"
else
warn "enckey not found in pod — run enckey-bootstrap.sh (or wait for PI to generate on first DB init)"
warn "enckey not found in pod — stop and follow privacyidea/KEY-RECOVERY.md; do not generate a replacement recovery key"
fi
if kubectl exec -n "$NAMESPACE" "$PI_POD" -- test -f /etc/privacyidea/private.pem 2>/dev/null; then
pass "audit private.pem present in pod"
else
warn "audit private.pem not found — run enckey-bootstrap.sh"
warn "audit private.pem not found — use privacyidea/KEY-RECOVERY.md for private export"
fi
if kubectl exec -n "$NAMESPACE" "$PI_POD" -- \

314
tests/test_privacyidea_key_custody.py Normal file
View file

@ -0,0 +1,314 @@
"""Key transport against real filesystem/crypto and a synthetic Kubernetes API."""
import base64
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
ROOT = Path(__file__).resolve().parents[1]
PI = ROOT / 'sso-mfa/k8s/privacyidea'
def module(name, file):
spec = importlib.util.spec_from_file_location(name, file)
result = importlib.util.module_from_spec(spec); spec.loader.exec_module(result)
return result
@pytest.fixture
def keys():
private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
return {'enckey': os.urandom(96),
'private.pem': private.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption()),
'public.pem': private.public_key().public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)}
@pytest.fixture
def case(tmp_path, keys):
root = tmp_path / 'private'; root.mkdir(mode=0o700)
(root / 'privacyidea').mkdir(mode=0o700)
volume = tmp_path / 'volume'; volume.mkdir(mode=0o750)
for name, value in keys.items():
(volume / name).write_bytes(value); (volume / name).chmod(0o600)
binpath = tmp_path / 'bin'; binpath.mkdir()
metadata = lambda name, uid: {'name': name, 'namespace': 'mfa', 'uid': uid, 'resourceVersion': '1'}
mount = {'name': 'data', 'mountPath': '/etc/privacyidea'}
spec = {'securityContext': {'fsGroup': 999}, 'volumes': [{'name': 'data', 'persistentVolumeClaim': {'claimName': 'privacyidea-data'}}],
'containers': [{'name': 'privacyidea', 'volumeMounts': [mount]}]}
deployment = {'apiVersion': 'apps/v1', 'kind': 'Deployment', 'metadata': {**metadata('privacyidea', 'deployment-uid'), 'generation': 1},
'spec': {'replicas': 1, 'template': {'spec': spec}}, 'status': {'replicas': 1, 'observedGeneration': 1}}
pod = {'apiVersion': 'v1', 'kind': 'Pod', 'metadata': {**metadata('privacyidea-source', 'pod-uid'),
'ownerReferences': [{'kind': 'ReplicaSet', 'name': 'privacyidea-rs', 'uid': 'rs-uid', 'controller': True}]},
'spec': spec, 'status': {'phase': 'Running', 'containerStatuses': [{'name': 'privacyidea', 'ready': True}]}}
state = {'deployment': deployment, 'pvc': {'apiVersion': 'v1', 'kind': 'PersistentVolumeClaim', 'metadata': metadata('privacyidea-data', 'pvc-uid'), 'status': {'phase': 'Bound'}},
'pods': {'apiVersion': 'v1', 'kind': 'PodList', 'items': [pod]},
'replicaset': {'apiVersion': 'apps/v1', 'kind': 'ReplicaSet', 'metadata': {**metadata('privacyidea-rs', 'rs-uid'),
'ownerReferences': [{'kind': 'Deployment', 'uid': 'deployment-uid', 'controller': True}]}}, 'secrets': {}}
statepath = tmp_path / 'state.json'; statepath.write_text(json.dumps(state))
binary = binpath / 'kubectl'
binary.write_text('''#!/usr/bin/python3
import base64,importlib.util,json,os,sys
from pathlib import Path
args=sys.argv[1:];p=Path(os.environ['STATE']);state=json.loads(p.read_text())
raw=sys.stdin.buffer.read() if args[0] in ('exec','create','patch') else b''
with open(os.environ['CALLS'],'a') as f:f.write(json.dumps({'args':args,'stdin':raw.decode()})+'\\n')
if os.environ.get('FAIL')==args[0]:
print('SYNTHETIC_SECRET_DIAGNOSTIC');print('SYNTHETIC_SECRET_DIAGNOSTIC',file=sys.stderr);sys.exit(1)
if args[0]=='get':
if args[1]=='secret':
obj=state['secrets'].get(args[2]);print(json.dumps(obj) if obj is not None else '');sys.exit(0)
if args[1]=='pod':print('privacyidea-source');sys.exit(0)
print(json.dumps(state[args[1]]));sys.exit(0)
if args[0]=='exec':
if args[-2]=='cat':sys.stdout.buffer.write((Path(os.environ['VOLUME'])/Path(args[-1]).name).read_bytes());sys.exit(0)
if 'test' in args:sys.exit(0)
spec=importlib.util.spec_from_file_location('remote',Path(os.environ['SOURCE'])/'key-custody-remote.py')
remote=importlib.util.module_from_spec(spec);spec.loader.exec_module(remote)
assert args[-1]==(Path(os.environ['SOURCE'])/'key-custody-remote.py').read_text()
request=json.loads(raw)
result=remote.operate(request,Path(os.environ['VOLUME']),uid=os.getuid(),gid=os.getgid())
if os.environ.get('RACE')=='pod':
state['pods']['items'][0]['metadata']['uid']='changed-pod';p.write_text(json.dumps(state))
print(json.dumps(result));sys.exit(0)
if args[0]=='create':
if 'generic' in args:print('{}');sys.exit(0)
obj=json.loads(raw);name=obj['metadata']['name']
if name in state['secrets'] or os.environ.get('RACE')=='create':sys.exit(1)
obj['metadata'].update(uid='secret-'+name,resourceVersion='1')
elif args[0]=='patch':
name=args[2];obj=state['secrets'][name]
if os.environ.get('RACE')=='patch':sys.exit(1)
for operation in json.loads(raw):
parts=[v.replace('~1','/').replace('~0','~') for v in operation['path'].split('/')[1:]];parent=obj
for part in parts[:-1]:parent=parent[part]
if operation['op']=='test':assert parent[parts[-1]]==operation['value']
elif operation['op']=='remove':del parent[parts[-1]]
else:sys.exit(2)
obj['metadata']['resourceVersion']='2'
elif args[0]=='apply':sys.exit(0)
else:sys.exit(2)
state['secrets'][name]=obj;p.write_text(json.dumps(state))
if os.environ.get('BAD_RESPONSE'):obj['data']['unexpected']='eA=='
print(json.dumps(obj))
'''); binary.chmod(0o700)
env = {'PATH': str(binpath) + ':' + os.environ['PATH'], 'HOME': str(tmp_path), 'STATE': str(statepath),
'CALLS': str(tmp_path/'calls'), 'VOLUME': str(volume), 'SOURCE': str(PI)}
return {'root': root, 'volume': volume, 'keys': keys, 'env': env, 'statepath': statepath}
def state(case): return json.loads(case['statepath'].read_text())
def save(case, value): case['statepath'].write_text(json.dumps(value))
def calls(case):
p=Path(case['env']['CALLS']);return [json.loads(line) for line in p.read_text().splitlines()] if p.exists() else []
def invoke(case, action='export', *args, **extra):
script = 'enckey-bootstrap.sh' if action=='export' else 'enckey-restore.sh'
return subprocess.run(['bash','-x',str(PI/script),str(case['root']),*args],env={**case['env'],**extra},capture_output=True,text=True,timeout=30)
def quiet(result, case):
assert 'SYNTHETIC_SECRET_DIAGNOSTIC' not in result.stdout+result.stderr
assert 'BEGIN RSA PRIVATE KEY' not in result.stdout+result.stderr
assert base64.b64encode(case['keys']['enckey']).decode() not in result.stdout+result.stderr
assert 'Traceback' not in result.stdout+result.stderr
def test_export_roundtrip_and_preserves_matching_copies(case):
result=invoke(case);assert result.returncode==0,result.stderr;quiet(result,case)
for name,value in case['keys'].items():
p=case['root']/'privacyidea'/('pi.enc' if name=='enckey' else name)
assert p.read_bytes()==value and p.stat().st_mode & 0o077==0
expected=state(case);count=len(calls(case));assert invoke(case).returncode==0
assert state(case)==expected and all(c['args'][0] in ('get','exec') for c in calls(case)[count:])
argv=json.dumps([c['args'] for c in calls(case)])
assert base64.b64encode(case['keys']['enckey']).decode() not in argv
assert all(c['args'][0]!='apply' for c in calls(case))
@pytest.mark.parametrize('defect',['different','symlink','hardlink','public','fifo'])
def test_export_refuses_bad_local_copy_before_secret_writes(case,defect):
p=case['root']/'privacyidea/pi.enc'
if defect=='fifo':os.mkfifo(p,0o600)
elif defect=='symlink':p.symlink_to(case['volume']/'enckey')
else:
p.write_bytes(b'old-copy' if defect=='different' else case['keys']['enckey']);p.chmod(0o644 if defect=='public' else 0o600)
if defect=='hardlink':os.link(p,p.with_name('alias'))
result=invoke(case);assert result.returncode!=0;quiet(result,case)
assert not state(case)['secrets'] and not (p.parent/'private.pem').exists()
if defect=='different':assert p.read_bytes()==b'old-copy'
@pytest.mark.parametrize('defect',['missing','short','link','mismatched-rsa'])
def test_export_refuses_invalid_remote_keys_without_generation(case,defect,keys):
p=case['volume']/'enckey'
if defect=='missing':p.unlink()
elif defect=='short':p.write_bytes(b'short')
elif defect=='link':os.link(p,p.with_name('alias'))
else:
other=rsa.generate_private_key(public_exponent=65537,key_size=2048)
(case['volume']/'public.pem').write_bytes(other.public_key().public_bytes(serialization.Encoding.PEM,serialization.PublicFormat.SubjectPublicKeyInfo))
result=invoke(case);assert result.returncode!=0;quiet(result,case)
assert not list((case['root']/'privacyidea').iterdir()) and not state(case)['secrets']
assert not any('pi-manage' in c['args'] for c in calls(case))
@pytest.mark.parametrize('defect',['two-pods','not-ready','wrong-owner','wrong-pvc','pod-race'])
def test_export_binds_ready_owned_pod_and_pvc(case,defect):
obj=state(case);pod=obj['pods']['items'][0]
if defect=='two-pods':obj['pods']['items'].append(pod)
elif defect=='not-ready':pod['status']['containerStatuses'][0]['ready']=False
elif defect=='wrong-owner':obj['replicaset']['metadata']['ownerReferences'][0]['uid']='other'
elif defect=='wrong-pvc':pod['spec']['volumes'][0]['persistentVolumeClaim']['claimName']='other'
save(case,obj)
result=invoke(case,RACE='pod' if defect=='pod-race' else '');assert result.returncode!=0;quiet(result,case)
assert not list((case['root']/'privacyidea').iterdir()) and not state(case)['secrets']
def test_export_failure_retains_local_recovery_copies(case):
result=invoke(case,FAIL='create');assert result.returncode!=0;quiet(result,case)
assert (case['root']/'privacyidea/pi.enc').read_bytes()==case['keys']['enckey']
assert not list((case['root']/'privacyidea').glob('.key-export-*'))
@pytest.mark.parametrize('defect',['different','legacy-field','owner','immutable','create-race','bad-response'])
def test_export_existing_backup_and_race_refusal(case,defect):
if defect not in ('create-race','bad-response'):
assert invoke(case).returncode==0
obj=state(case);secret=obj['secrets']['privacyidea-enckey']
if defect=='different':secret['data']['enckey']=base64.b64encode(b'other').decode()
elif defect=='legacy-field':secret['data']['PI_ENCFILE']=secret['data'].pop('enckey')
elif defect=='owner':secret['metadata']['ownerReferences']=[{'uid':'other-owner'}]
else:secret['immutable']=True
save(case,obj)
before=state(case)
result=invoke(case,RACE='create' if defect=='create-race' else '',BAD_RESPONSE='yes' if defect=='bad-response' else '')
assert result.returncode!=0;quiet(result,case)
if defect!='bad-response':assert state(case)==before
def test_export_removes_annotation_conditionally_and_preserves_extra_data(case):
assert invoke(case).returncode==0;obj=state(case)
for secret in obj['secrets'].values():
secret['metadata']['annotations']={'kubectl.kubernetes.io/last-applied-configuration':'synthetic-copy','owner':'keep'}
secret['data']['extra']='eA=='
save(case,obj);assert invoke(case).returncode==0
for secret in state(case)['secrets'].values():
assert secret['metadata']['annotations']=={'owner':'keep'} and secret['data']['extra']=='eA=='
patches=[json.loads(c['stdin']) for c in calls(case) if c['args'][0]=='patch']
assert len(patches)==2 and all({op['path'] for op in patch if op['op']=='test'}=={'/metadata/uid','/metadata/resourceVersion','/data'} for patch in patches)
def test_explicit_first_bootstrap_creates_only_missing_audit_pair(case):
for name in ('private.pem','public.pem'):(case['volume']/name).unlink()
assert invoke(case).returncode!=0
result=invoke(case,'export','--initialize-audit-keys');assert result.returncode==0,result.stderr;quiet(result,case)
assert (case['volume']/'enckey').read_bytes()==case['keys']['enckey']
assert invoke(case,'export','--initialize-audit-keys').returncode!=0
assert invoke(case).returncode==0
def recovery(case):
assert invoke(case).returncode==0
host=module('host',PI/'key-custody.py')
pod=host.render('ghcr.io/gpappsoft/privacyidea-docker@sha256:'+'a'*64,'pvc-uid')
pod['metadata'].update(uid='restore-pod-uid',resourceVersion='1')
pod['status']={'phase':'Running','containerStatuses':[{'name':'recovery','ready':True}]}
obj=state(case);obj['pods']['items']=[pod];obj['deployment']['spec']['replicas']=0;obj['deployment']['status']['replicas']=0;save(case,obj)
for name in case['keys']:(case['volume']/name).unlink()
def test_restore_installs_on_actual_volume_and_retry_preserves_files(case):
recovery(case);result=invoke(case,'restore','--pvc-uid','pvc-uid');assert result.returncode==0,result.stderr;quiet(result,case)
for name,value in case['keys'].items():assert (case['volume']/name).read_bytes()==value
before={p.name:p.stat().st_ino for p in case['volume'].iterdir()}
assert invoke(case,'restore','--pvc-uid','pvc-uid').returncode==0
assert before=={p.name:p.stat().st_ino for p in case['volume'].iterdir()}
@pytest.mark.parametrize('defect',['deployment-running','pod-terminating','other-pod','wrong-pvc-uid','mutable-image','no-mount','app-entrypoint','token-mount'])
def test_restore_rejects_unstopped_or_wrong_target_before_key_transfer(case,defect):
recovery(case);obj=state(case);pod=obj['pods']['items'][0]
if defect=='deployment-running':obj['deployment']['spec']['replicas']=1
elif defect=='pod-terminating':pod['metadata']['deletionTimestamp']='now'
elif defect=='other-pod':obj['pods']['items'].append(pod)
elif defect=='wrong-pvc-uid':obj['pvc']['metadata']['uid']='replacement-pvc'
elif defect=='mutable-image':pod['spec']['containers'][0]['image']='image:latest'
elif defect=='no-mount':pod['spec']['containers'][0]['volumeMounts']=[]
elif defect=='app-entrypoint':pod['spec']['containers'][0]['command']=['pi-manage','run']
else:pod['spec']['automountServiceAccountToken']=True
save(case,obj);count=len(calls(case));result=invoke(case,'restore','--pvc-uid','pvc-uid')
assert result.returncode!=0;quiet(result,case)
assert not list(case['volume'].iterdir()) and all(c['args'][0]=='get' for c in calls(case)[count:])
def test_restore_refuses_partial_mismatching_state_without_overwrite(case):
recovery(case);p=case['volume']/'private.pem';p.write_bytes(b'other');p.chmod(0o600)
result=invoke(case,'restore','--pvc-uid','pvc-uid');assert result.returncode!=0;quiet(result,case)
assert p.read_bytes()==b'other' and not (case['volume']/'enckey').exists()
def test_restored_key_decrypts_existing_native_privacyidea_values(case):
pytest.importorskip('privacyidea')
from privacyidea.lib.security.default import DefaultSecurityModule
source=DefaultSecurityModule({'file':str(case['volume']/'enckey')})
ciphertext=source.encrypt_pin('synthetic-token-pin')
password=source.encrypt_password('synthetic-configuration-value')
from privacyidea.lib.crypto import Sign
signer=Sign((case['volume']/'private.pem').read_bytes(),(case['volume']/'public.pem').read_bytes())
signature=signer.sign('synthetic-audit-record')
recovery(case);assert invoke(case,'restore','--pvc-uid','pvc-uid').returncode==0
restored=DefaultSecurityModule({'file':str(case['volume']/'enckey')})
assert restored.decrypt_pin(ciphertext)=='synthetic-token-pin'
assert restored.decrypt_password(password)=='synthetic-configuration-value'
recovered_signer=Sign((case['volume']/'private.pem').read_bytes(),(case['volume']/'public.pem').read_bytes())
assert recovered_signer.verify('synthetic-audit-record',signature)
assert signer.verify('synthetic-new-audit-record',recovered_signer.sign('synthetic-new-audit-record'))
def test_remote_partial_restore_keeps_keys_and_retries_without_replacement(tmp_path, keys, monkeypatch):
remote=module('partial_remote',PI/'key-custody-remote.py')
real_link=remote.os.link
def interrupted(source,target,**kwargs):
if target=='public.pem':raise OSError('synthetic-write-failure')
return real_link(source,target,**kwargs)
request={'action':'restore','keys':{k:base64.b64encode(v).decode() for k,v in keys.items()}}
monkeypatch.setattr(remote.os,'link',interrupted)
with pytest.raises(OSError):remote.operate(request,tmp_path,uid=os.getuid(),gid=os.getgid())
assert (tmp_path/'enckey').read_bytes()==keys['enckey']
assert (tmp_path/'private.pem').read_bytes()==keys['private.pem']
assert not list(tmp_path.glob('.key-custody-*'))
inodes={p.name:p.stat().st_ino for p in tmp_path.iterdir()}
monkeypatch.setattr(remote.os,'link',real_link)
assert remote.operate(request,tmp_path,uid=os.getuid(),gid=os.getgid())=={'result':'restored'}
assert all((tmp_path/name).stat().st_ino==inode for name,inode in inodes.items())
def test_remote_partial_audit_initialization_cannot_generate_another_pair(tmp_path, keys, monkeypatch):
remote=module('init_remote',PI/'key-custody-remote.py')
p=tmp_path/'enckey';p.write_bytes(keys['enckey']);p.chmod(0o600)
real_link=remote.os.link
def interrupted(source,target,**kwargs):
if target=='public.pem':raise OSError('synthetic-write-failure')
return real_link(source,target,**kwargs)
monkeypatch.setattr(remote.os,'link',interrupted)
with pytest.raises(OSError):remote.operate({'action':'initialize-audit'},tmp_path,uid=os.getuid(),gid=os.getgid())
before=(tmp_path/'private.pem').read_bytes()
monkeypatch.setattr(remote.os,'link',real_link)
with pytest.raises(remote.Refused):remote.operate({'action':'initialize-audit'},tmp_path,uid=os.getuid(),gid=os.getgid())
assert (tmp_path/'private.pem').read_bytes()==before
def test_unknown_argument_is_not_echoed(case):
result=invoke(case,'export','--SYNTHETIC_SECRET_DIAGNOSTIC')
assert result.returncode!=0;quiet(result,case);assert not calls(case)
def test_kubectl_generic_list_output_is_accepted(case):
obj=state(case);obj['pods']['kind']='List';save(case,obj)
assert invoke(case).returncode==0

42
workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md
View file

@ -196,11 +196,10 @@ or live recipient migration is claimed. Validation and exact source binding:
Custodian `docs/evidence/2026-10-13-netkingdom-security/sops-write-callers.json`.
Remaining closure checklist (source review, not proof of current live use):
- `privacyidea/enckey-bootstrap.sh` is still called by `creds-bootstrap-agent.sh`
and documented by `creds-apply.sh` and the privacyIDEA README. It exports keys to
ordinary overwriteable paths and uses client-side Secret apply. Repair private
no-overwrite export, existing-key preservation and conditional DR Secret writes;
verify the actual backup/restore contract before recovery acceptance.
- `privacyidea/enckey-bootstrap.sh` and its bootstrap/documentation callers are
repaired below, including private no-overwrite export and a real PVC recovery
transport. Actual off-host backup lineage, reviewed image/cluster execution and
native service/database/token/audit recovery acceptance remain T03 requirements.
- `privacyidea/check-user-mfa-state.sh`, linked by the smooth-bootstrap guide,
still reads retired `keycape-pi-token`, disables TLS verification, puts its token
in curl argv and prints provider error text. Reconcile with the current scoped
@ -221,6 +220,39 @@ Remaining closure checklist (source review, not proof of current live use):
- After source coverage, execute T03's exact owner-approved off-host recovery and
current-consumer checks. Source regressions do not satisfy that gate.
2026-10-13 privacyIDEA key export/recovery source repair: `key-custody.py`
and its captured in-pod adapter verify the Ready deployment-owned source pod and
Bound PVC, validate the 96-byte encryption key and matching RSA audit pair, and
retain private local copies without overwrite. DR Secrets are create-or-verify;
existing extra fields/metadata survive and annotation removal uses conditional
UID/version/data tests. The shared Secret publisher now accepts binary values.
Missing encryption keys are never regenerated. Audit-pair creation is a separate
explicit first-bootstrap flag, refused when any local/cluster recovery copy or
partial pair already exists. The bootstrap caller forwards that flag only when
explicitly selected, and script errors/key bytes are not forwarded to the terminal.
Restore now has an executable captured transport to the actual `privacyidea-data`
PVC. A rendered helper uses an independently reviewed immutable image digest,
no application entrypoint or service-account mount, and the real data volume.
The wrapper requires a stopped Deployment, exact PVC UID and the sole Ready
mounting helper pod, refuses differing files and retains partial writes for
reconciliation. Files are published without replacement and checked afterward;
this is not a multi-file transaction or UID-atomic Kubernetes exec. The admitted
new-file ownership is root:999 with mode 0640 for private files, matching the
source fsGroup. Current runtime ownership/consumer acceptance still needs proof.
`privacyidea/KEY-RECOVERY.md` replaces the broken unmounted-helper and Secret-only
restart examples, resolves `enckey` versus `PI_ENCFILE`, and keeps stopping,
creating/removing the reviewed helper and resuming the service explicit in the
owner ceremony. Disposable native privacyIDEA 3.12.2 checks prove decryption of
pre-recovery values and audit signature continuity after the local restore.
Failure, partial-publication, wrong-target and mismatch tests plus broader results
are retained in Custodian
`docs/evidence/2026-10-13-netkingdom-security/privacyidea-key-custody.json`.
No production key, PVC, workload or Secret was read or changed for these tests.
T02 remains progress for the remaining diagnostic/LDAP/realm callers and full
inventory/lifecycle reconciliation. T03 remains the native recovery gate.
## Verify owner-approved recovery and current native callers
```task