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

NK-WP-0049-T42: rebuild the MFA stage around the restored directory; check by user and realm
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details

Browse source 
The founder's first real run stopped at ResolverError on all checks: real tokens
belong to an LDAP user and the archived resolver names the in-cluster directory.
Restored LLDAP now joins the database's network-free namespace, archived resolver
addresses are repointed to loopback in the disposable copy only (any other address
is refused), and the checks call POST /validate/check by user and realm as KeyCape
does. Fifteen native fixture tests pass (HOTP and TOTP+PIN, negatives, bundle).
Also covers the user-to-token resolver and the encrypted bind password.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 328925@bnt-lap001
Assistant-Session: 14b795fd-c632-4c01-8d28-e7e490c7b0ed
This commit is contained in:
tegwick 2026-10-24 18:42:35 +02:00
parent 9a3a84c99c
commit 5008e34f1a
6 changed files with 365 additions and 200 deletions
Download patch file Download diff file Expand all files Collapse all files

70
docs/identity-service-restore-packet.md
View file

@ -265,41 +265,53 @@ procedure) and your authenticator ready:
bash tools/bootstrap-recovery/run-isolated-privacyidea-mfa.sh /path/to/verified/transfer.json
```
Prompts, in order: the `railiance-backup` age key (hidden); a token serial chosen
from the list printed on your terminal (serial, type, usable flag; the list never
enters the receipt); then that token's PIN plus current code exactly as typed at
login (hidden). Pick a token you hold and whose PIN you know. A TOTP code is valid
Prompts, in order: the `railiance-backup` age key (hidden); your login name (and
the realm, if more than one exists); then a token PIN plus current code exactly as
typed at login (hidden). Any one of your usable tokens works. A TOTP code is valid
for tens of seconds, so enter it only when asked and keep the authenticator open.
The restore is the privacyIDEA stage's: restored database and original keys in
`--network none` containers, read-only, capabilities dropped, user 1000, tmpfs
only, isolation inspected before private import. Three checks then run through
privacyIDEA's own `check_serial_pass`, each in a separate process:
**Why by user.** A first version checked a token by serial. On the real backup every
check stopped with `ResolverError`: real tokens belong to an LDAP user, and the
archived resolver `lldap-coulomb` points at the in-cluster address
`ldap://lldap.sso.svc.cluster.local:3890`, which does not exist in isolation. Fixture
tokens had no owner, so tests had not shown it, and the run correctly claimed no proof.
1. **Wrong key.** The application is pointed at a random key file; the run asserts
the configuration really names it. The typed code must be refused (native
decryption raises). This is the discriminating step: the same code is accepted
with the original key in step 3.
2. **Wrong code.** The final digit of what you typed is changed; it must be refused.
3. **Your code.** Accepted by the original key and retained token material.
**Topology.** The restored database container is the `--network none` anchor.
Restored LLDAP and the privacyIDEA worker join its namespace (`container:<anchor>`)
and talk over loopback only. All three are read-only, capabilities dropped, user
non-root, tmpfs-only, with isolation inspected before private import.
The order is deliberate. Only the last step can spend a counter or replay-protected
code. A first fixture run showed why: when the wrong-key step silently used the
original key, it accepted the code and the real check was then refused. The
configuration assertion now prevents that vacuous result, and the receipt only
claims proof when all three outcomes line up.
**Resolver rewrite, disposable copy only.** After `pg_restore`, every `ldapresolver`
`LDAPURI` must equal the archived in-cluster address exactly, or the run is refused
as an owner decision. They are then repointed to `ldap://127.0.0.1:3890`. The stored
bind password is not touched, so the original encryption key must decrypt it and the
restored LLDAP must accept it before any user can be looked up.
The receipt carries booleans, bounded refusal categories, token type, a SHA-256 of
the serial, counts and source hashes only. No serial, PIN or code is recorded, and
the disposable copy is destroyed. Fixture tests cover HOTP and TOTP-with-PIN, a
wrong code alone, an unknown serial, a malformed original key and an encrypted
bundle through `main()`.
**Checks.** The call is `POST /validate/check` with user, realm and pass, through
privacyIDEA's own application, so stored policies and the resolver apply as for KeyCape:
**Not proven here:** the user-to-token path. KeyCape calls `/validate/check` with
user, realm and pass, so the restored LDAP resolver, its encrypted bind password
and the LLDAP lookup are exercised only by a chained run (T18 phase B). A policy
such as `otppin=userstore` would not apply to a serial-based check. The receipt
therefore records `user_to_token_resolver_proven: false`.
1. The user is found through the restored resolver and directory, and has a usable token.
2. **Wrong key.** The application is pointed at a random key file (the run asserts the
configuration really names it). The typed code must be refused; privacyIDEA returns
its own error 907.
3. **Wrong code.** The last digit of what you typed is changed; refused as a wrong value.
4. **Your code.** Accepted with the original key and the retained token material.
The order is deliberate: only the last step can spend a counter or replay-protected
code. An earlier fixture run in which the wrong-key step silently used the original key
showed why; the configuration assertion now prevents that vacuous result, and the
receipt claims proof only when all three outcomes line up.
The receipt carries booleans, bounded refusal categories (for example `error_907`,
`wrong_value`), token types, counts, a SHA-256 of the login and source hashes. No login,
PIN or code is recorded, and the disposable copy is destroyed. Fixture tests cover HOTP
and TOTP-with-PIN owned by an LDAP user, a wrong code alone, an unknown user, a malformed
original key, the rewrite refusing an unknown address, and an encrypted bundle through
`main()`.
**Limits.** Policies that call outside the namespace (for example event handlers that
send notifications) cannot succeed and are not exercised. Token ownership is checked for
the one user you name, not for every user.
## Unresolved inputs before execution

193
tests/test_isolated_privacyidea_mfa.py
View file

@ -1,34 +1,82 @@
"""Existing-token MFA on a restored synthetic privacyIDEA database (disposable fixtures only)."""
"""Existing-token MFA by user and realm on a restored synthetic privacyIDEA + LLDAP pair."""
import importlib.util
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
import pytest
from test_identity_bundle import stage # noqa: F401 (fixture reuse)
from test_isolated_lldap_recovery import PASSWORD, retained_directory # noqa: F401 (fixture reuse)
from test_isolated_privacyidea_recovery import retained_state # noqa: F401 (fixture reuse)
ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('pi_mfa', ROOT / 'tools/bootstrap-recovery/isolated-privacyidea-mfa.py')
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)
BASE = 'dc=example,dc=test'
CODE = r'''
import contextlib,io,json,logging
SETUP = r'''
import contextlib,io,json,logging,sys
logging.disable(logging.CRITICAL)
v=json.load(sys.stdin)
with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()):
from privacyidea.app import create_app
from privacyidea.lib.token import get_tokens
from privacyidea.lib.resolver import save_resolver
from privacyidea.lib.realm import set_realm,set_default_realm
from privacyidea.lib.user import User
from privacyidea.lib.token import init_token,assign_token,get_tokens
app=create_app(config_name='testing',config_file='/recovery/pi.cfg',silent=True)
with app.app_context():
out=[]
for t in get_tokens():
res,pin,otp,_=t.get_otp()
out.append({'serial':t.token.serial,'otp':otp})
print(json.dumps(out))
save_resolver({'resolver':'lldap-coulomb','type':'ldapresolver','LDAPURI':'ldap://127.0.0.1:3890',
'LDAPBASE':'ou=people,'+v['base'],'BINDDN':'uid=admin,ou=people,'+v['base'],'BINDPW':v['password'],
'LOGINNAMEATTRIBUTE':'uid','LDAPSEARCHFILTER':'(uid=*)','LDAPFILTER':'(&(uid=%s))',
'USERINFO':json.dumps({'username':'uid','email':'mail'}),
'UIDTYPE':'uid','SCOPE':'SUBTREE','AUTHTYPE':'Simple','TIMEOUT':'5','CACHE_TIMEOUT':'0',
'SIZELIMIT':'500','NOREFERRALS':'True','TLS_VERIFY':'False'})
set_realm('coulomb',[{'name':'lldap-coulomb','priority':1}])
set_default_realm('coulomb')
owner=User('admin','coulomb')
assert owner.exist()
if v['kind']=='totp':
token=init_token({'type':'totp','genkey':1,'pin':'synthetic-pin','serial':'TOTPFIXTURE01'})
assign_token('TOTPFIXTURE01',owner)
serial='TOTPFIXTURE01'
else:
serial=get_tokens()[0].token.serial
assign_token(serial,owner)
otp=get_tokens(serial=serial)[0].get_otp()[2]
print(json.dumps({'serial':serial,'otp':otp}))
'''
@pytest.fixture(params=['hotp', 'totp'])
def owned_state(request, tmp_path, retained_state, retained_directory): # noqa: F811
"""Dump of a database whose resolver names the in-cluster address, as production does."""
root, database, worker = retained_state
shutil.copytree(tmp_path / 'data/lldap', root / 'data/lldap')
shutil.copytree(tmp_path / 'resources/sso', root / 'resources/sso')
directory = database + '-fixture-dir'
try:
m.start_directory(directory, database, root)
m.run(['docker', 'exec', worker, 'python3', '-c', m.WAIT_LDAP], timeout=90)
raw = m.run(['docker', 'exec', '-i', worker, 'python3', '-c', SETUP],
input=json.dumps({'base': BASE, 'password': PASSWORD, 'kind': request.param}).encode(), timeout=120)
made = json.loads(raw)
assert m.psql(database, "UPDATE resolverconfig SET \"Value\"='" + m.CLUSTER_LDAP + "' WHERE \"Key\"='LDAPURI'") == 'UPDATE 1'
(root / 'databases/privacyidea.dump').write_bytes(m.run(
['docker', 'exec', database, 'pg_dump', '-U', 'postgres', '-Fc', 'recovery']))
yield root, made['otp'], request.param
finally:
m.pi.cleanup([directory])
def passcode(otp, kind):
return ('synthetic-pin' if kind == 'totp' else '') + otp
def test_launcher_preflight_entry_point_exists_and_launcher_calls_it():
# The launcher swallows exceptions into one message; a missing attribute once hid here.
assert callable(m.preflight)
@ -37,8 +85,6 @@ def test_launcher_preflight_entry_point_exists_and_launcher_calls_it():
def test_launcher_preflight_snippet_passes_against_the_pinned_target():
import re
import subprocess
launcher = (ROOT / 'tools/bootstrap-recovery/run-isolated-privacyidea-mfa.sh').read_text()
snippet = re.search(r"python3 - <<'PY'\n(.*?)\nPY\n", launcher, re.S).group(1)
done = subprocess.run(['python3', '-c', snippet], cwd=ROOT, capture_output=True, text=True, timeout=120)
@ -51,70 +97,94 @@ def test_wrong_code_keeps_shape_but_changes_value():
assert len(m.different_code('pin123456')) == len('pin123456')
def test_serial_pattern_refuses_shell_and_path_characters():
for bad in ('', 'a b', '../x', 'a;b', 'x' * 65):
assert not m.SERIAL.fullmatch(bad)
assert m.SERIAL.fullmatch('HOTP0001A2B3')
def test_login_and_realm_patterns_refuse_shell_and_path_characters():
for bad in ('', 'a b', '../x', 'a;b', 'x' * 129):
assert not m.LOGIN.fullmatch(bad)
assert m.LOGIN.fullmatch('first.last@example.test')
assert not m.REALM.fullmatch('a/b') and m.REALM.fullmatch('coulomb')
def retained_codes(worker):
return {x['serial']: x['otp'] for x in json.loads(m.run(['docker', 'exec', worker, 'python3', '-c', CODE]))}
def test_resolver_rewrite_accepts_only_the_archived_cluster_address(monkeypatch):
calls = []
def fake(database, statement):
calls.append(statement)
return 'UPDATE 2' if statement.startswith('UPDATE') else '\n'.join([m.CLUSTER_LDAP] * 2)
monkeypatch.setattr(m, 'psql', fake)
assert m.rewrite_resolvers('db') == 2 and calls[1].startswith('UPDATE')
monkeypatch.setattr(m, 'psql', lambda d, s: 'ldap://elsewhere.example:389')
with pytest.raises(m.bundle.Refused):
m.rewrite_resolvers('db')
monkeypatch.setattr(m, 'psql', lambda d, s: '')
with pytest.raises(m.bundle.Refused):
m.rewrite_resolvers('db')
monkeypatch.setattr(m, 'psql', lambda d, s: 'UPDATE 1' if s.startswith('UPDATE') else m.CLUSTER_LDAP + '\n' + m.CLUSTER_LDAP)
with pytest.raises(m.bundle.Refused):
m.rewrite_resolvers('db')
def test_native_existing_token_passes_and_negatives_are_refused(retained_state):
root, _, worker = retained_state
codes = retained_codes(worker)
serial, otp = sorted(codes.items())[0]
def test_native_existing_token_passes_by_user_and_negatives_are_refused(owned_state):
root, otp, kind = owned_state
result = {}
m.restore_and_check(root, result, lambda tokens: (serial, otp))
m.restore_and_check(root, result, lambda realms, default: ('admin', default), lambda summary: passcode(otp, kind))
assert result['existing_token_mfa_proven'] is True, result
assert result['wrong_key_refused'] and result['wrong_code_refused']
assert result['existing_token_native_check_accepted']
assert result['retained_tokens_inventoried'] == 2 and result['selected_token_type'] == 'hotp'
assert result['wrong_key_refused'] and result['wrong_code_refused'] and result['existing_token_native_check_accepted']
assert result['user_to_token_resolver_proven'] and result['user_resolved_through_restored_resolver_and_directory']
assert result['ldap_resolvers_rewritten_to_loopback'] == 1 and result['user_usable_token_types'] == [kind]
assert result['container_cleanup_verified'] and result['container_isolation_verified_before_private_import']
assert result['only_loopback_interface_verified'] and result['external_routes_absent']
text = json.dumps(result)
assert serial not in text and otp not in text
assert otp not in text and 'admin' not in text.replace('"selected_user_sha256"', '')
def test_native_wrong_code_alone_never_proves_mfa(retained_state):
root, _, worker = retained_state
serial, otp = sorted(retained_codes(worker).items())[0]
def test_native_wrong_code_alone_never_proves_mfa(owned_state):
root, otp, kind = owned_state
result = {}
with pytest.raises(m.bundle.Refused):
m.restore_and_check(root, result, lambda tokens: (serial, m.different_code(otp)))
m.restore_and_check(root, result, lambda realms, default: ('admin', default),
lambda summary: m.different_code(passcode(otp, kind)))
assert result['failed_stage'] == 'acceptance'
assert result['existing_token_mfa_proven'] is False and not result['existing_token_native_check_accepted']
assert result['container_cleanup_verified']
def test_native_unknown_serial_is_refused_before_any_check(retained_state):
root, _, _ = retained_state
def test_native_unknown_user_is_refused_before_any_code_is_requested(owned_state):
root, _, _ = owned_state
result = {}
asked = []
with pytest.raises(m.bundle.Refused):
m.restore_and_check(root, result, lambda tokens: ('NOSUCHTOKEN', '123456'))
assert result['failed_stage'] == 'owner_selection'
assert 'wrong_key_refused' not in result and result['container_cleanup_verified']
m.restore_and_check(root, result, lambda realms, default: ('nosuchuser', default),
lambda summary: asked.append(1) or '123456')
assert result['failed_stage'] == 'user_lookup_through_restored_resolver' and not asked
assert result['user_resolved_through_restored_resolver_and_directory'] is False
assert result['container_cleanup_verified']
def test_native_malformed_original_key_does_not_prove_mfa(retained_state):
root, _, worker = retained_state
serial, otp = sorted(retained_codes(worker).items())[0]
def test_native_malformed_original_key_does_not_prove_mfa(owned_state):
root, otp, kind = owned_state
(root / 'data/privacyidea/enckey').write_bytes(b'synthetic-malformed-key')
result = {}
with pytest.raises(m.bundle.Refused):
m.restore_and_check(root, result, lambda tokens: (serial, otp))
assert result['existing_token_mfa_proven'] is False and result['container_cleanup_verified']
m.restore_and_check(root, result, lambda realms, default: ('admin', default), lambda summary: passcode(otp, kind))
assert not result.get('existing_token_mfa_proven') and not result.get('existing_token_native_check_accepted')
assert result['failed_stage'] and result['container_cleanup_verified']
def test_encrypted_bundle_to_native_mfa_and_tmpfs_cleanup(retained_state, stage, tmp_path, monkeypatch, capsys): # noqa: F811
import shutil
def test_encrypted_bundle_to_native_mfa_and_tmpfs_cleanup(owned_state, stage, tmp_path, monkeypatch, capsys): # noqa: F811
import sys
root, _, worker = retained_state
serial, otp = sorted(retained_codes(worker).items())[0]
root, otp, kind = owned_state
for key in ('enckey', 'private.pem', 'public.pem'):
shutil.copyfile(root / 'data/privacyidea' / key, stage / 'data/privacyidea' / key)
shutil.copyfile(root / 'databases/privacyidea.dump', stage / 'databases/privacyidea.dump')
for source in (root / 'data/lldap').iterdir():
if source.is_file():
shutil.copyfile(source, stage / 'data/lldap' / source.name)
for kind_, name in (('deployment', 'lldap'), ('secret', 'lldap-secrets')):
target = stage / 'resources/sso' / kind_ / (name + '.json')
value = json.loads(target.read_text())
value.update({k: v for k, v in json.loads((root / 'resources/sso' / kind_ / (name + '.json')).read_text()).items()
if k != 'metadata'})
target.write_text(json.dumps(value))
resource = stage / 'resources/mfa/secret/privacyidea-config.json'
value = json.loads(resource.read_text())
value['data'] = json.loads((root / 'resources/mfa/secret/privacyidea-config.json').read_text())['data']
@ -138,42 +208,15 @@ def test_encrypted_bundle_to_native_mfa_and_tmpfs_cleanup(retained_state, stage,
'ciphertext_sha256': m.local.offline.offsite.digest(ciphertext), 'etag': '"synthetic"',
'fetched_path': str(ciphertext), 'independent_owner_retrieval_verified': True}))
monkeypatch.setattr(m.local.verify.custody, 'private_key_fd', lambda: os.open(key, os.O_RDONLY))
monkeypatch.setattr(m, 'owner_choice', lambda tokens: (serial, otp))
# main() binds restore_and_check's callback by name at call time.
monkeypatch.setattr(m, 'owner_user', lambda realms, default: ('admin', default))
monkeypatch.setattr(m, 'owner_code', lambda summary: passcode(otp, kind))
receipt = tmp_path / 'receipt.json'
monkeypatch.setattr(sys, 'argv', ['isolated-privacyidea-mfa', '--transfer-receipt', str(transfer),
'--receipt', str(receipt)])
assert m.main() == 0, receipt.read_text()
result = json.loads(receipt.read_text())
assert result['status'] == 'isolated_privacyidea_mfa_verified' and result['existing_token_mfa_proven']
assert result['user_to_token_resolver_proven'] is False
assert result['user_to_token_resolver_proven'] is True
assert result['container_cleanup_verified'] and result['temporary_plaintext_cleanup_verified']
assert not result['production_contacted'] and not result['secret_values_exported']
text = capsys.readouterr().out + receipt.read_text()
assert serial not in text and otp not in text
ADD_TOTP = r'''
import contextlib,io,json,logging
logging.disable(logging.CRITICAL)
with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()):
from privacyidea.app import create_app
from privacyidea.lib.token import init_token
app=create_app(config_name='testing',config_file='/recovery/pi.cfg',silent=True)
with app.app_context():
token=init_token({'type':'totp','genkey':1,'pin':'synthetic-pin','serial':'TOTPFIXTURE01'})
print(json.dumps({'serial':'TOTPFIXTURE01'}))
'''
def test_native_totp_token_with_pin_passes_within_the_check_sequence(retained_state):
root, database, worker = retained_state
m.run(['docker', 'exec', worker, 'python3', '-c', ADD_TOTP])
(root / 'databases/privacyidea.dump').write_bytes(m.run(
['docker', 'exec', database, 'pg_dump', '-U', 'postgres', '-Fc', 'recovery']))
code = retained_codes(worker)['TOTPFIXTURE01']
result = {}
m.restore_and_check(root, result, lambda tokens: ('TOTPFIXTURE01', 'synthetic-pin' + code))
assert result['selected_token_type'] == 'totp' and result['retained_tokens_inventoried'] == 3
assert result['existing_token_mfa_proven'] is True, result
assert result['wrong_code_refusal'] in ('wrong_value', 'wrong_pin', 'other_refusal')
assert otp not in capsys.readouterr().out + receipt.read_text()

12
tools/bootstrap-recovery/README.md
View file

@ -38,12 +38,12 @@ wrong-password and wrong-client-secret refusal. Notifier, NTP and healthcheck
writes are confined; nothing leaves loopback. It imports `isolated-lldap.py`
unedited. Synthetic native tests and the actual owner run (2026-10-24) passed; T17 is done.
`run-isolated-privacyidea-mfa.sh TRANSFER_RECEIPT` is the existing-token MFA step (T19,
prepared under T42). After the age key it shows a token list on the terminal, then asks
for a serial and the PIN+code you would type at login. `isolated-privacyidea-mfa.py` checks
a wrong key, a wrong code and your code through privacyIDEA's own validation on the
restored database. It reuses `isolated-privacyidea.py` unedited and does not prove the
user-to-token resolver. Fixture tests pass; the actual owner run is still open.
`run-isolated-privacyidea-mfa.sh TRANSFER_RECEIPT` is the existing-token MFA step (T19, prepared under
T42). After the age key it asks for your login name, then the PIN+code you would type at login.
`isolated-privacyidea-mfa.py` restores LLDAP, the database and the original keys into one network-free
namespace, repoints the archived LDAP resolver to the restored directory in the disposable copy only, and calls
`/validate/check` by user and realm, as KeyCape does: wrong key refused, wrong code refused, your code accepted.
It reuses `isolated-privacyidea.py` and `isolated-lldap.py` unedited. Fixture tests pass; the actual owner run is still open.
`run-isolated-privacyidea.sh TRANSFER_RECEIPT` is the token-material/audit recovery step.
It asks only for the existing backup age key. `isolated-privacyidea.py` restores

230
tools/bootstrap-recovery/isolated-privacyidea-mfa.py
View file

@ -1,10 +1,11 @@
#!/usr/bin/env python3
"""Owner-run native privacyIDEA existing-token MFA check on the restored database.
The owner selects one retained token and types its PIN+OTP privately. The restored
token material is checked by privacyIDEA's own validation code inside a loopback-only
disposable container: first with a wrong encryption key, then with a wrong code, then
with the code as typed. Serials and codes never enter receipts, argv or environment.
Restored LLDAP, the restored database and a privacyIDEA worker share one network-free
namespace. The owner names their login and types a token's PIN+OTP privately. The call
is POST /validate/check by user and realm, as KeyCape makes it, through the restored LDAP
resolver: first with a wrong encryption key, then with a wrong code, then as typed.
Codes never enter receipts, argv or environment; the login is stored only as a hash.
"""
import argparse
import hashlib
@ -26,27 +27,57 @@ spec = importlib.util.spec_from_file_location('isolated_privacyidea', HERE / 'is
pi = importlib.util.module_from_spec(spec)
spec.loader.exec_module(pi)
local, bundle, run = pi.local, pi.bundle, pi.run
SERIAL = re.compile(r'[A-Za-z0-9_.:-]{1,64}')
LOGIN = re.compile(r'[A-Za-z0-9_.@-]{1,128}')
REALM = re.compile(r'[A-Za-z0-9_.-]{1,64}')
CLUSTER_LDAP = 'ldap://lldap.sso.svc.cluster.local:3890'
LOOPBACK_LDAP = 'ldap://127.0.0.1:3890'
INVENTORY = r'''
REALMS = r"""
import contextlib,io,json,logging
logging.disable(logging.CRITICAL)
result={'status':'failed'}
try:
with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()):
from privacyidea.app import create_app
from privacyidea.models import Token
from privacyidea.lib.realm import get_realms,get_default_realm
app=create_app(config_name='testing',config_file='/recovery/pi.cfg',silent=True)
with app.app_context():
tokens=[{'serial':t.serial,'type':t.tokentype,'active':bool(t.active),'locked':bool(t.locked),
'revoked':bool(t.revoked),'otplen':t.otplen} for t in Token.query.order_by(Token.serial).limit(200)]
result={'status':'ok','tokens':tokens}
names=sorted(get_realms());default=get_default_realm()
result={'status':'ok','realms':names,'default':default}
except Exception as error:result={'status':'failed','failure_type':type(error).__name__}
print(json.dumps(result))
'''
"""
# Looks the user up through the restored resolver and the restored directory.
RESOLVE = r"""
import contextlib,io,json,logging,sys
logging.disable(logging.CRITICAL)
v=json.load(sys.stdin)
result={'status':'failed'}
stage='native_application_configuration'
try:
with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()):
from privacyidea.app import create_app
app=create_app(config_name='testing',config_file='/recovery/pi.cfg',silent=True)
with app.app_context():
from privacyidea.lib.user import User
from privacyidea.lib.token import get_tokens
stage='user_lookup_through_restored_resolver'
user=User(v['login'],v['realm'])
found=bool(user.exist())
tokens=[]
if found:
stage='user_token_listing'
tokens=[{'type':t.type,'active':bool(t.is_active()),'locked':bool(t.token.locked),
'revoked':bool(t.token.revoked)} for t in get_tokens(user=user)]
result={'status':'ok','user_found':found,'tokens':tokens}
except Exception as error:result={'status':'failed','failed_stage':stage,'failure_type':type(error).__name__}
print(json.dumps(result))
"""
# One check per process: privacyIDEA loads its encryption module once per app.
CHECK = r'''
# The call is the endpoint KeyCape uses, so stored policies and the resolver both apply.
CHECK = r"""
import contextlib,io,json,logging,os,sys
logging.disable(logging.CRITICAL)
v=json.load(sys.stdin)
@ -73,25 +104,34 @@ try:
with app.app_context():
# A wrong-key run that silently used the original key would be vacuous.
assert app.config['PI_ENCFILE']==('/recovery/wrong-enckey' if v['mode']=='wrong_key' else '/recovery/enckey')
from privacyidea.lib.token import check_serial_pass
from privacyidea.models import Token
token=Token.query.filter_by(serial=v['serial']).first()
assert token is not None
stage='native_validation'
accepted=False;refusal='none'
accepted=False;refusal='none';code=None
try:
accepted,reply=check_serial_pass(v['serial'],v['passcode'],options={})
accepted=bool(accepted)
message=str((reply or {}).get('message','')).lower()
refusal='none' if accepted else ('wrong_pin' if 'pin' in message else
'wrong_value' if ('otp' in message or 'wrong' in message) else 'other_refusal')
response=app.test_client().post('/validate/check',data={'user':v['login'],'realm':v['realm'],'pass':v['passcode']})
body=response.get_json(silent=True) or {}
outcome=body.get('result') or {}
accepted=response.status_code==200 and outcome.get('status') is True and outcome.get('value') is True
code=(outcome.get('error') or {}).get('code')
message=str((body.get('detail') or {}).get('message','')).lower()
if accepted:refusal='none'
elif code is not None:refusal='error_'+str(code)
else:refusal='wrong_pin' if 'pin' in message else 'wrong_value' if ('otp' in message or 'wrong' in message) else 'other_refusal'
except Exception as error:
accepted=False;refusal='exception_'+type(error).__name__[:40]
result={'status':'completed','accepted':accepted,'refusal':refusal,'token_type':token.tokentype,
result={'status':'completed','accepted':accepted,'refusal':refusal,
'only_loopback_interface_verified':True,'external_routes_absent':True}
except Exception as error:result={'status':'failed','failed_stage':stage,'failure_type':type(error).__name__}
print(json.dumps(result))
'''
"""
WAIT_LDAP = r"""
import socket,sys,time
for _ in range(240):
try:
with socket.create_connection(('127.0.0.1',3890),timeout=1):sys.exit(0)
except OSError:time.sleep(.25)
sys.exit(1)
"""
def preflight():
@ -111,54 +151,118 @@ def different_code(passcode):
return passcode[:-1] + ('0' if last != '0' else '1')
def restore_and_check(root, receipt, ask):
"""ask(inventory) -> (serial, passcode) after the containers are contained."""
def psql(database, statement):
out = run(['docker', 'exec', database, 'psql', '-U', 'postgres', '-d', 'recovery', '-v', 'ON_ERROR_STOP=1',
'-At', '-c', statement])
return out.decode().strip()
def rewrite_resolvers(database):
"""Point archived LDAP resolvers at the restored directory, in the disposable copy only.
Only the archived in-cluster address is accepted; anything else is an owner decision.
The encrypted bind password is left untouched so the original key must decrypt it.
"""
rows = psql(database, 'SELECT c."Value" FROM resolverconfig c JOIN resolver r ON r.id=c.resolver_id '
"WHERE r.rtype='ldapresolver' AND c.\"Key\"='LDAPURI'").splitlines()
if not rows or any(value != CLUSTER_LDAP for value in rows):
raise bundle.Refused()
changed = psql(database, 'UPDATE resolverconfig SET "Value"=\'' + LOOPBACK_LDAP + '\' WHERE "Key"=\'LDAPURI\' '
"AND resolver_id IN (SELECT id FROM resolver WHERE rtype='ldapresolver') "
"AND \"Value\"='" + CLUSTER_LDAP + "'")
if changed != 'UPDATE ' + str(len(rows)):
raise bundle.Refused()
return len(rows)
def start_directory(name, database, root):
"""Restored LLDAP joins the database's network-free namespace (loopback only)."""
config, jwt, admin_password, _ = local.configuration(root)
database_id = run(['docker', 'container', 'inspect', database, '--format', '{{.Id}}']).decode().strip()
run(['docker', 'run', '--detach', '--pull=never', '--name', name,
'--network', 'container:' + database_id, '--read-only', '--cap-drop', 'ALL',
'--security-opt', 'no-new-privileges', '--user', '1000:1000',
'--memory', '256m', '--cpus', '1', '--pids-limit', '128',
'--tmpfs', '/data:rw,nosuid,nodev,noexec,size=268435456,uid=1000,gid=1000,mode=0700',
'--tmpfs', '/run/recovery:rw,nosuid,nodev,noexec,size=1048576,uid=1000,gid=1000,mode=0700',
'--tmpfs', '/tmp:rw,nosuid,nodev,noexec,size=16777216,uid=1000,gid=1000',
'--env', 'LLDAP_JWT_SECRET_FILE=/run/recovery/jwt',
'--env', 'LLDAP_LDAP_USER_PASS_FILE=/run/recovery/admin-password',
'--entrypoint', '/bin/sh', local.IMAGE, '-c', 'sleep 1800'])
pi.isolation(name, 'container:' + database_id, '1000:1000')
payload = {p.relative_to(root / 'data/lldap').as_posix(): p.read_bytes()
for p in (root / 'data/lldap').rglob('*') if p.is_file()}
run(['docker', 'exec', '-i', name, 'tar', '-xf', '-', '-C', '/data'], input=local.archive_files(payload))
run(['docker', 'exec', '-i', name, 'tar', '-xf', '-', '-C', '/run/recovery'],
input=local.archive_files({'lldap.toml': config, 'jwt': jwt, 'admin-password': admin_password}))
run(['docker', 'exec', '--detach', name, '/bin/sh', '-c',
'exec /app/lldap run --config-file /run/recovery/lldap.toml > /dev/null 2>&1'])
def restore_and_check(root, receipt, choose_user, choose_code):
"""choose_user(realms, default) -> (login, realm); choose_code(summary) -> PIN+code."""
stem = 'nk49-pi-mfa-' + uuid.uuid4().hex
database, worker = stem + '-db', stem + '-worker'
database, directory, worker = stem + '-db', stem + '-dir', stem + '-worker'
stage = 'database_restore'
try:
pi.start_database(database)
with (root / 'databases/privacyidea.dump').open('rb') as dump:
run(['docker', 'exec', '-i', database, 'pg_restore', '-U', 'postgres', '-d', 'recovery',
'--no-owner', '--no-privileges', '--exit-on-error'], stdin=dump, timeout=600)
stage = 'resolver_rewrite'
receipt['ldap_resolvers_rewritten_to_loopback'] = rewrite_resolvers(database)
stage = 'directory_launch'
start_directory(directory, database, root)
stage = 'worker_launch'
pi.start_worker(worker, database, root)
receipt['container_isolation_verified_before_private_import'] = True
stage = 'token_inventory'
inventory = worker_json(worker, INVENTORY)
if inventory.get('status') != 'ok' or not inventory['tokens']:
stage = 'directory_startup'
run(['docker', 'exec', worker, 'python3', '-c', WAIT_LDAP], timeout=90)
stage = 'realm_listing'
realms = worker_json(worker, REALMS)
if realms.get('status') != 'ok' or not realms['realms']:
raise bundle.Refused()
tokens = {t['serial']: t for t in inventory['tokens']}
receipt['retained_tokens_inventoried'] = len(tokens)
stage = 'owner_selection'
serial, passcode = ask(inventory['tokens'])
if serial not in tokens or not passcode or len(passcode) > 256:
stage = 'owner_user_selection'
login, realm = choose_user(realms['realms'], realms.get('default'))
if not LOGIN.fullmatch(login) or realm not in realms['realms']:
raise bundle.Refused()
chosen = tokens[serial]
if not chosen['active'] or chosen['locked'] or chosen['revoked']:
stage = 'user_lookup_through_restored_resolver'
user = worker_json(worker, RESOLVE, {'login': login, 'realm': realm})
if user.get('status') != 'ok':
receipt['failed_stage'] = user.get('failed_stage', stage)
raise bundle.Refused()
receipt['user_resolved_through_restored_resolver_and_directory'] = bool(user['user_found'])
if not user['user_found']:
raise bundle.Refused()
usable = [t for t in user['tokens'] if t['active'] and not t['locked'] and not t['revoked']]
receipt.update(selected_user_sha256=hashlib.sha256(login.encode()).hexdigest(),
user_token_count=len(user['tokens']), user_usable_token_count=len(usable),
user_usable_token_types=sorted({t['type'] for t in usable}))
stage = 'owner_code_entry'
if not usable:
raise bundle.Refused()
passcode = choose_code({'tokens': user['tokens'], 'usable': len(usable)})
if not passcode or len(passcode) > 256:
raise bundle.Refused()
receipt.update(selected_token_serial_sha256=hashlib.sha256(serial.encode()).hexdigest(),
selected_token_type=chosen['type'])
steps = (('wrong_key', passcode), ('wrong_code', different_code(passcode)), ('correct', passcode))
outcomes = {}
for mode, code in steps:
for mode, code in (('wrong_key', passcode), ('wrong_code', different_code(passcode)), ('correct', passcode)):
stage = 'check_' + mode
outcome = worker_json(worker, CHECK, {'mode': mode, 'serial': serial, 'passcode': code})
outcome = worker_json(worker, CHECK, {'mode': mode, 'login': login, 'realm': realm, 'passcode': code})
if outcome.get('status') != 'completed':
receipt['failed_stage'] = outcome.get('failed_stage', stage)
raise bundle.Refused()
outcomes[mode] = outcome
receipt.update(
wrong_key_refused=not outcomes['wrong_key']['accepted'],
wrong_key_refusal=outcomes['wrong_key']['refusal'],
wrong_code_refused=not outcomes['wrong_code']['accepted'],
wrong_code_refusal=outcomes['wrong_code']['refusal'],
wrong_key_refused=not outcomes['wrong_key']['accepted'], wrong_key_refusal=outcomes['wrong_key']['refusal'],
wrong_code_refused=not outcomes['wrong_code']['accepted'], wrong_code_refusal=outcomes['wrong_code']['refusal'],
existing_token_native_check_accepted=outcomes['correct']['accepted'],
only_loopback_interface_verified=all(o['only_loopback_interface_verified'] for o in outcomes.values()),
external_routes_absent=all(o['external_routes_absent'] for o in outcomes.values()))
# Order matters: a replay-protected code is spent only by the last, accepted check.
# Only the last step can spend a counter or replay-protected code.
proven = (receipt['wrong_key_refused'] and receipt['wrong_code_refused']
and receipt['existing_token_native_check_accepted'])
receipt['existing_token_mfa_proven'] = proven
receipt['user_to_token_resolver_proven'] = proven
if not proven:
stage = 'acceptance'
raise bundle.Refused()
@ -166,25 +270,25 @@ def restore_and_check(root, receipt, ask):
receipt.setdefault('failed_stage', stage)
raise
finally:
pi.cleanup([worker, database])
pi.cleanup([worker, directory, database])
receipt['container_cleanup_verified'] = True
def owner_choice(tokens):
"""Terminal-only: serials are shown to the owner, never written to a receipt."""
from_tty = sys.stdin.isatty() and sys.stderr.isatty()
if not from_tty:
def owner_user(realms, default):
"""Terminal-only prompts; the login is hashed in receipts."""
if not (sys.stdin.isatty() and sys.stderr.isatty()):
raise bundle.Refused()
print('Retained tokens (shown only on this terminal):')
for t in tokens:
flags = ','.join(x for x, on in (('inactive', not t['active']), ('locked', t['locked']),
('revoked', t['revoked'])) if on) or 'usable'
print(' %-24s %-8s otplen=%s %s' % (t['serial'], t['type'], t['otplen'], flags))
serial = input('Serial of the token whose code you will enter: ').strip()
if not SERIAL.fullmatch(serial):
raise bundle.Refused()
passcode = local.offline.private_prompt('Token PIN + current code, as typed at login (hidden): ')
return serial, passcode
print('Realms in the restored database: ' + ', '.join(realms) + (' (default: %s)' % default if default else ''))
login = input('Your login name (the user your token belongs to): ').strip()
realm = default or realms[0]
if len(realms) > 1:
realm = input('Realm [%s]: ' % realm).strip() or realm
return login, realm
def owner_code(summary):
print('Found %d token(s) for that user, %d usable. Any one of them may be used.' % (len(summary['tokens']), summary['usable']))
return local.offline.private_prompt('Token PIN + current code, as typed at login (hidden): ')
def main():
@ -236,7 +340,7 @@ def main():
result.update(ciphertext_sha256=transfer['ciphertext_sha256'],
manifest_file_count=len(manifest['files']), all_manifest_hashes_verified=True)
stage = 'native_mfa_check'
restore_and_check(restored, result, owner_choice)
restore_and_check(restored, result, owner_user, owner_code)
if work.exists():
raise bundle.Refused()
result.update(status='isolated_privacyidea_mfa_verified', temporary_plaintext_cleanup_verified=True)

10
tools/bootstrap-recovery/run-isolated-privacyidea-mfa.sh
View file

@ -25,10 +25,10 @@ print('OK: exact bnt-lap001 recovery engine, pinned images and tmpfs available.'
PY
NK49_PI_RUN="$(mktemp -d /tmp/nk49-isolated-privacyidea-mfa-XXXXXXXX)"
printf 'Safe evidence directory: %s\n' "$NK49_PI_RUN"
printf '%s\n' 'This restores the linked database and original keys only into disposable network-isolated containers.'
printf '%s\n' 'Prompts, in order: the backup age key (hidden), then a token serial from the list shown on this terminal,'
printf '%s\n' 'then that token PIN+current code as you type it at login (hidden). Have the authenticator ready first.'
printf '%s\n' 'Checks: wrong key refused, wrong code refused, your code accepted. The restored copy is disposable; production is untouched.'
printf '%s\n' 'This restores the privacyIDEA database, its original keys and the LLDAP directory into disposable network-free containers.'
printf '%s\n' 'Prompts, in order: the backup age key (hidden), your login name, then a token PIN+current code as typed at login (hidden).'
printf '%s\n' 'Have your authenticator open first. Checks: wrong key refused, wrong code refused, your code accepted, by user and realm.'
printf '%s\n' 'The restored copy is disposable; production is untouched.'
if bash sso-mfa/bootstrap/sops-custody-unlock.sh \
--expected-recipient age1zvryunvjhvpkmasskauga2heeg0ztnte9ymgppvjge36ekumk50syr3tsz \
--prompt -- python3 tools/bootstrap-recovery/isolated-privacyidea-mfa.py \
@ -39,4 +39,4 @@ else
printf 'Check stopped or needs review. Safe receipt: %s/recovery.json\n' "$NK49_PI_RUN"
exit "$NK49_PI_EXIT"
fi
printf '%s\n' 'Return only the outcome and evidence directory. The user-to-token resolver and the KeyCape consumer run remain separate.'
printf '%s\n' 'Return only the outcome and evidence directory. The KeyCape consumer run remains separate.'

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

@ -567,7 +567,7 @@ status: wait
priority: high
assignee: codex
depends_on: [NK-WP-0049-T15, NK-WP-0049-T16, NK-WP-0049-T31, NK-WP-0049-T42]
blocking_reason: "LLDAP T16, token-material T31 and audit T20 passed; T42 stage prepared and fixture-tested 2026-10-24. Await the founder-attended run: run-isolated-privacyidea-mfa.sh with a verified transfer receipt, the age key and a token PIN+code."
blocking_reason: "LLDAP T16, token-material T31 and audit T20 passed; T42 stage prepared and fixture-tested 2026-10-24 and rebuilt after the first run (resolver/directory now restored; checks run by user and realm). Await the founder-attended run: run-isolated-privacyidea-mfa.sh with a verified transfer receipt, the age key, a login name and a token PIN+code."
state_hub_task_id: "6f8af1a7-2a3f-5fa8-8202-83103ebd55b2"
```
@ -1550,29 +1550,35 @@ state_hub_task_id: "2fa3bd7c-12a9-546a-8098-a1999346da8a"
**Accountable worker:** claude-code; the founder holds the token, its PIN and the
age key, and runs the T19 stage.
Completed 2026-10-24. `isolated-privacyidea-mfa.py`, `run-isolated-privacyidea-mfa.sh`
and `tests/test_isolated_privacyidea_mfa.py` add the T19 stage: the privacyIDEA
stage's restore and containment helpers (reused unedited), a terminal-only token list,
and three native `check_serial_pass` runs in separate processes (wrong key, wrong
code, owner's code). Eight opt-in native fixture tests pass on the pinned images,
including HOTP, TOTP with a PIN, a wrong code alone, an unknown serial, a malformed
original key and an encrypted-bundle run through `main()`.
Completed 2026-10-24, then redesigned the same day after the founder's first real run.
`isolated-privacyidea-mfa.py`, `run-isolated-privacyidea-mfa.sh` and
`tests/test_isolated_privacyidea_mfa.py` add the T19 stage. Restored LLDAP, the
database and a privacyIDEA worker share one network-free namespace; the archived LDAP
resolver address is repointed to the restored directory in the disposable copy only;
and three native `POST /validate/check` calls by user and realm run in separate
processes (wrong key, wrong code, the owner's code), the call KeyCape makes. Reuses
`isolated-privacyidea.py` and `isolated-lldap.py` unedited. Fifteen opt-in native fixture
tests pass on the pinned images: HOTP and TOTP-with-PIN tokens owned by an LDAP user,
a wrong code alone, an unknown user, a malformed original key, the rewrite refusing an
unknown resolver address, and an encrypted-bundle run through `main()`.
A first fixture run was vacuous and the acceptance gate caught it: privacyIDEA
applies `PRIVACYIDEA_CONFIGFILE` after the `config_file` argument, so the wrong-key
run used the original key, accepted the code and spent the HOTP counter. The probe
now sets the variable to the file in use and asserts the intended `PI_ENCFILE`.
Two defects were found and fixed, both before any acceptance claim. (1) A first fixture
run was vacuous: privacyIDEA applies `PRIVACYIDEA_CONFIGFILE` after the `config_file`
argument, so the wrong-key run used the original key, accepted the code and spent the
HOTP counter. The probe now sets the variable and asserts the intended `PI_ENCFILE`.
(2) The launcher called `m.preflight()` on a module that did not define it, and its
catch-all printed the generic preflight message; tests now run the launcher's own
snippet. Then the founder's first real run (receipt retained in `/tmp`, not in Git)
stopped at acceptance with `ResolverError` on all three checks: the first design checked
by token serial, but real tokens belong to an LDAP user whose resolver points at the
in-cluster `lldap.sso.svc.cluster.local`, absent in isolation. The stage was rebuilt
around the directory and a by-user check; the gate had correctly refused to claim proof.
No key, PIN or code was recorded, and nothing was consumed.
Residual, recorded in the receipt as `user_to_token_resolver_proven: false`: KeyCape
validates by user and realm, so the restored LDAP resolver, its encrypted bind
password and the LLDAP user lookup need a chained run (T18 phase B). This prepares
T19; it is not T19 acceptance and touches no production service.
Correction 2026-10-24, found when the founder first ran the launcher: it called
`m.preflight()` on a module that only exposed `pi.preflight`, and its catch-all printed
the generic preflight message. The fixture tests had called `main()` directly and
never went through the launcher. Fixed; tests now execute the launcher's own
preflight snippet. No key was requested and nothing was consumed.
This prepares T19 and also covers the user-to-token resolver, its encrypted bind
password and the LLDAP user lookup (receipt field `user_to_token_resolver_proven`). It
is not T19 acceptance and touches no production service. Event handlers or policies that
call outside the namespace are not exercised.
## Accept remaining recovery and caller evidence and close the G08 handoff