NK-WP-0049-T41: isolated Authelia restore stage, fixture-tested
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

isolated-authelia.py runs restored LLDAP as the network-none anchor with
Authelia joining its namespace. Checks the original config, storage key,
issuer key, KeyCape's client secret and existing-user first factor, with
wrong-password and wrong-client-secret refusal. Confinement rewrites: LDAP
to loopback, file notifier, NTP check off, healthcheck file off. Imports
isolated-lldap.py unedited; the T38-pinned KeyCape sources are untouched.
Native fixture tests on the pinned images pass (9). T17 now awaits the
founder-attended run.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 328925@bnt-lap001
Assistant-Session: 14b795fd-c632-4c01-8d28-e7e490c7b0ed
This commit is contained in:
tegwick 2026-10-24 17:14:52 +02:00
parent 4719970abb
commit 006dc23fdf
7 changed files with 783 additions and 3 deletions

View file

@ -290,7 +290,7 @@
| task | NK-WP-0049-T38 | done | — | workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md |
| task | NK-WP-0049-T39 | done | — | workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md |
| task | NK-WP-0049-T40 | done | — | workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md |
| task | NK-WP-0049-T41 | progress | — | workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md |
| task | NK-WP-0049-T41 | done | — | workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md |
| task | NK-WP-0050-T01 | done | — | workplans/NK-WP-0050-playwright-consumer-registration.md |
| task | NK-WP-0050-T02 | progress | — | workplans/NK-WP-0050-playwright-consumer-registration.md |
| task | NK-WP-0050-T03 | wait | — | workplans/NK-WP-0050-playwright-consumer-registration.md |

View file

@ -130,6 +130,60 @@ For future runs, return only the outcome and evidence directory.
This step does not restore Authelia, KeyCape or privacyIDEA, and does not prove
current production password validity, MFA or audit continuity.
## Prepared owner step: isolated Authelia recovery (T17)
Prepared and fixture-tested 2026-10-24 under T41; not yet run on the actual
backup. Execute in your own terminal with a verified transfer receipt from an
independent retrieval (the T23 procedure), an existing account and the password
valid at the backup's capture time:
```bash
bash tools/bootstrap-recovery/run-isolated-authelia.sh EXISTING_USER \
/path/to/verified/transfer.json
```
The command asks privately for the `railiance-backup` age key and then the
password. It rechecks ciphertext size/hash, recipient and every archive file,
and decrypts into private Linux tmpfs, as the LLDAP step does.
Topology: restored LLDAP runs as the `--network none` anchor. Authelia and the
probe client join its network namespace (`container:<anchor>`) and use loopback
only. All containers are `nk49-authelia-*`, read-only, capabilities dropped, no
added privileges, user 1000, tmpfs-only mounts, no published ports. Isolation
is inspected before private import; the probe verifies there is no external
interface or route.
The archived deployment, ConfigMap and `authelia-secrets` are used as captured.
Secret files are placed at their archived mount paths; the environment carries
only file paths. Four confinement rewrites are made, and only these: the LDAP
address points to the restored directory on loopback; the SMTP notifier becomes
a tmpfs file notifier and `authelia-smtp` is not imported; the NTP startup check
is off (outbound UDP); the container healthcheck file is off (the root stays
read-only). KeyCape's archived `keycape-config` supplies the client id, client
secret and the portal hostname.
Checks, in order: `authelia validate-config` on the original configuration;
`authelia storage encryption check` with the original key; the issuer JWKS
served by Authelia contains the archived issuer public key; existing-user
first factor through `/api/firstfactor` against the restored LLDAP; a wrong
password refused; the archived `keycape` client secret passes client
authentication at `/api/oidc/token` (`invalid_grant` for a deliberately invalid
code, so no grant is issued); a wrong client secret refused (`invalid_client`).
The receipt carries booleans, the RFC 7638 issuer key thumbprint, image IDs and
source hashes only. Containers and tmpfs staging are removed and checked on
every exit.
Native fixture tests (`NK49_RUN_ISOLATED_LLDAP_TEST=1
NK49_RUN_ISOLATED_AUTHELIA_TEST=1 pytest tests/test_isolated_authelia_recovery.py`)
use this repository's own Authelia manifests with synthetic secrets on the
pinned images. They cover the positive path, wrong password, wrong storage key
and an encrypted-bundle run through `main()`.
This step does not prove the full KeyCape→Authelia→LLDAP→privacyIDEA login,
MFA, or current production password validity. It reuses `isolated-lldap.py`
and the bundle modules unedited, so the pinned KeyCape attempt sources stay as
reviewed.
## Acceptance matrix
| Task / owner lane | Positive check | Negative check | Safe evidence |

View file

@ -0,0 +1,242 @@
"""Isolated Authelia restoration with synthetic retained credentials only."""
import base64
import importlib.util
import json
import os
from pathlib import Path
import secrets
import subprocess
import uuid
import pytest
import yaml
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from test_identity_bundle import stage # noqa: F401 (fixture reuse)
from test_isolated_lldap_recovery import PASSWORD, retained_directory # noqa: F401 (fixture reuse)
import test_isolated_lldap_recovery as lldap_fixture
ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('isolated_authelia', ROOT / 'tools/bootstrap-recovery/isolated-authelia.py')
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)
MANIFESTS = ROOT / 'sso-mfa/k8s/authelia'
CLIENT_SECRET = 'synthetic-keycape-client-' + secrets.token_hex(16) # Disposable fixture only.
def secret_values():
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
return {'jwt_secret': secrets.token_hex(32), 'session_secret': secrets.token_hex(32),
'storage_encryption_key': secrets.token_hex(32), 'ldap_password': PASSWORD,
'oidc_hmac_secret': secrets.token_hex(32),
'oidc_issuer_private_key': key.private_bytes(serialization.Encoding.PEM,
serialization.PrivateFormat.TraditionalOpenSSL,
serialization.NoEncryption()).decode(),
'keycape_client_secret_hash': 'placeholder-replaced-in-native-fixture'}
def resources(root, values):
"""Write the repository's own Authelia manifests as archived resources."""
def write(kind, name, value):
p = root / 'resources/sso' / kind
p.mkdir(parents=True, exist_ok=True)
value = {**value, 'metadata': {**value.get('metadata', {}), 'name': name, 'namespace': 'sso'}}
(p / (name + '.json')).write_text(json.dumps(value))
deployment = next(d for d in yaml.safe_load_all((MANIFESTS / 'deployment.yaml').read_text())
if d and d['kind'] == 'Deployment')
configmap = yaml.safe_load((MANIFESTS / 'configmap.yaml').read_text())
config = yaml.safe_load(configmap['data']['configuration.yml'])
ldap = config['authentication_backend']['ldap']
ldap.update(base_dn='dc=example,dc=test', user='uid=admin,ou=people,dc=example,dc=test')
configmap['data']['configuration.yml'] = yaml.safe_dump(config, sort_keys=False)
write('deployment', 'authelia', deployment)
write('configmap', 'authelia-config', configmap)
write('secret', 'authelia-secrets', {'data': {k: base64.b64encode(v.encode()).decode() for k, v in values.items()}})
write('secret', 'authelia-smtp', {'data': {'password': base64.b64encode(b'synthetic-smtp-canary').decode()}})
keycape = {'authelia': {'clientId': 'keycape', 'clientSecret': CLIENT_SECRET,
'browserBaseURL': 'https://auth.coulomb.social'}}
write('secret', 'keycape-config', {'data': {'config.yaml': base64.b64encode(yaml.safe_dump(keycape).encode()).decode()}})
store = root / 'data/authelia/db.sqlite3'
store.parent.mkdir(parents=True, exist_ok=True)
if not store.exists():
store.write_bytes(b'')
def test_configuration_keeps_originals_and_confines_ldap_and_notifier(tmp_path):
values = secret_values()
resources(tmp_path, values)
runtime = m.configuration(tmp_path)
config = yaml.safe_load(runtime['config'])
assert config['authentication_backend']['ldap']['url'] == m.LDAP_URL
assert config['notifier'] == {'filesystem': {'filename': '/var/authelia/data/' + m.NOTIFICATIONS}}
assert config['ntp']['disable_startup_check'] is True
assert config['server']['disable_healthcheck'] is True
assert config['identity_providers']['oidc']['clients'][0]['secret'].startswith('{{ secret ')
assert runtime['secrets']['/run/secrets/authelia/storage_encryption_key'] == values['storage_encryption_key'].encode()
assert not any('smtp' in name.lower() for name in runtime['env'])
assert all(path.startswith('/run/secrets/authelia/') for name, path in runtime['env'].items()
if name.endswith('_FILE'))
assert runtime['host'] == 'auth.coulomb.social' and runtime['client_secret'] == CLIENT_SECRET
for value in values.values():
assert value.encode() not in runtime['config']
def test_configuration_refuses_missing_secret_file(tmp_path):
values = secret_values()
del values['storage_encryption_key']
resources(tmp_path, values)
with pytest.raises(m.bundle.Refused):
m.configuration(tmp_path)
def test_configuration_refuses_env_from_unknown_source(tmp_path):
resources(tmp_path, secret_values())
path = tmp_path / 'resources/sso/deployment/authelia.json'
value = json.loads(path.read_text())
value['spec']['template']['spec']['containers'][0]['env'].append(
{'name': 'AUTHELIA_EXTRA', 'valueFrom': {'secretKeyRef': {'name': 'other', 'key': 'x'}}})
path.write_text(json.dumps(value))
with pytest.raises(m.bundle.Refused):
m.configuration(tmp_path)
def test_configuration_refuses_unknown_keycape_client(tmp_path):
resources(tmp_path, secret_values())
path = tmp_path / 'resources/sso/secret/keycape-config.json'
keycape = {'authelia': {'clientId': 'not-registered', 'clientSecret': CLIENT_SECRET,
'browserBaseURL': 'https://auth.coulomb.social'}}
path.write_text(json.dumps({'metadata': {'name': 'keycape-config', 'namespace': 'sso'}, 'data': {'config.yaml': base64.b64encode(yaml.safe_dump(keycape).encode()).decode()}}))
with pytest.raises(m.bundle.Refused):
m.configuration(tmp_path)
def test_public_jwk_thumbprint_is_rfc7638():
pem = secret_values()['oidc_issuer_private_key'].encode()
jwk, thumbprint = m.public_jwk(pem)
assert set(jwk) == {'e', 'kty', 'n'} and len(thumbprint) == 43
@pytest.fixture
def retained_authelia(retained_directory): # noqa: F811
if os.environ.get('NK49_RUN_ISOLATED_AUTHELIA_TEST') != '1':
pytest.skip('explicit local disposable-image fixture opt-in')
m.preflight()
values = secret_values()
digest = m.command(['docker', 'run', '--rm', '--pull=never', '--network', 'none', '--entrypoint', '/app/authelia',
m.IMAGE, 'crypto', 'hash', 'generate', 'pbkdf2', '--variant', 'sha512',
'--password', CLIENT_SECRET]).decode()
values['keycape_client_secret_hash'] = digest.split('Digest: ', 1)[1].strip()
resources(retained_directory, values)
runtime = m.configuration(retained_directory)
(retained_directory / 'data/authelia/db.sqlite3').unlink()
name = 'nk49-authelia-fixture-' + uuid.uuid4().hex
try:
m.command(['docker', 'run', '--detach', '--pull=never', '--name', name, '--network', 'none',
'--read-only', '--cap-drop', 'ALL', '--user', '1000:1000',
*m.tmpfs('/var/authelia/data', 67108864), *m.tmpfs('/run/secrets/authelia', 1048576),
*m.tmpfs('/config', 1048576), *[x for k, v in runtime['env'].items() for x in ('--env', k + '=' + v)],
'--entrypoint', '/bin/sh', m.IMAGE, '-c', 'sleep 600'])
m.command(['docker', 'exec', '-i', name, 'tar', '-xf', '-', '-C', '/run/secrets/authelia'],
input=m.archive_files({k.rsplit('/', 1)[1]: v for k, v in runtime['secrets'].items()}))
m.command(['docker', 'exec', '-i', name, 'tar', '-xf', '-', '-C', '/config'],
input=m.archive_files({'configuration.yml': runtime['config']}))
m.command(['docker', 'exec', name, '/app/authelia', 'storage', 'migrate', 'up',
'--config', '/config/configuration.yml'])
store = m.command(['docker', 'exec', name, 'cat', '/var/authelia/data/db.sqlite3'])
(retained_directory / 'data/authelia/db.sqlite3').write_bytes(store)
yield retained_directory, values
finally:
subprocess.run(['docker', 'rm', '--force', '--volumes', name], capture_output=True, timeout=30)
def test_native_restored_authelia_passes_all_checks(retained_authelia):
root, values = retained_authelia
result = {}
m.restore_authelia(root, 'admin', PASSWORD, result)
assert result['status'] == 'verified', result
for key in ('original_configuration_validated', 'storage_encryption_key_verified', 'issuer_key_matches_archive',
'existing_user_first_factor', 'wrong_password_refused', 'archived_client_secret_accepted',
'wrong_client_secret_refused', 'only_loopback_interface_verified', 'container_cleanup_verified',
'container_isolation_verified_before_private_import', 'notifications_confined_to_tmpfs_file'):
assert result[key] is True, key
assert result['grant_issued'] is False
text = json.dumps(result)
assert PASSWORD not in text and CLIENT_SECRET not in text
assert not any(v in text for v in values.values())
def test_native_wrong_password_fails_and_removes_containers(retained_authelia):
root, _ = retained_authelia
result = {}
with pytest.raises(m.bundle.Refused):
m.restore_authelia(root, 'admin', 'wrong-retained-password', result)
assert result['failed_stage'] == 'existing_user_first_factor'
assert result['container_cleanup_verified']
def test_native_wrong_storage_key_is_refused(retained_authelia):
root, _ = retained_authelia
path = root / 'resources/sso/secret/authelia-secrets.json'
value = json.loads(path.read_text())
value['data']['storage_encryption_key'] = base64.b64encode(secrets.token_hex(32).encode()).decode()
path.write_text(json.dumps(value))
result = {}
with pytest.raises(m.bundle.Refused):
m.restore_authelia(root, 'admin', PASSWORD, result)
assert result['failed_stage'] == 'storage_decryption'
assert result['container_cleanup_verified']
assert lldap_fixture.m.IMAGE == m.lldap.IMAGE
def test_encrypted_bundle_to_native_authelia_and_tmpfs_cleanup(retained_authelia, stage, tmp_path, monkeypatch, capsys):
import shutil
import sys
root, values = retained_authelia
# Replace the synthetic placeholders with the retained LLDAP/Authelia state,
# keeping the complete profile's resource identities and other key assets.
for component in ('lldap', 'authelia'):
for source in (root / 'data' / component).iterdir():
if source.is_file():
shutil.copyfile(source, stage / 'data' / component / source.name)
for kind, name in (('deployment', 'lldap'), ('secret', 'lldap-secrets'), ('deployment', 'authelia'),
('configmap', 'authelia-config'), ('secret', 'authelia-secrets'), ('secret', 'authelia-smtp')):
target = stage / 'resources/sso' / kind / (name + '.json')
value = json.loads(target.read_text())
retained = json.loads((root / 'resources/sso' / kind / (name + '.json')).read_text())
value.update({k: v for k, v in retained.items() if k != 'metadata'})
target.write_text(json.dumps(value))
target = stage / 'resources/sso/secret/keycape-config.json'
value = json.loads(target.read_text())
value['data']['config.yaml'] = json.loads(
(root / 'resources/sso/secret/keycape-config.json').read_text())['data']['config.yaml']
target.write_text(json.dumps(value))
key = tmp_path / 'fixture-age.key'
m.command(['age-keygen', '-o', str(key)])
recipient = m.command(['age-keygen', '-y', str(key)]).decode().strip()
profile = m.bundle.profile()
profile['public_recipient'] = recipient
monkeypatch.setattr(m.bundle, 'profile', lambda: profile)
artifact = tmp_path / 'net-kingdom-identity-20261016T063159Z-0123456789ab.tar.age'
m.bundle.seal(stage, artifact, {'synthetic_native_fixture': True})
transfer = tmp_path / 'transfer.json'
transfer.write_text(json.dumps({'schema': 'netkingdom.identity-offsite.v1', 'status': 'readback_verified',
'provider': 'Storage Share', 'folder': 'Backup/railiance-backups/net-kingdom-identity',
'artifact': artifact.name, 'ciphertext_bytes': artifact.stat().st_size,
'ciphertext_sha256': m.offline.offsite.digest(artifact), 'etag': '"synthetic"',
'fetched_path': str(artifact), 'independent_owner_retrieval_verified': True}))
monkeypatch.setattr(m.verify.custody, 'private_key_fd', lambda: os.open(key, os.O_RDONLY))
monkeypatch.setattr(m.offline, 'private_prompt', lambda label: PASSWORD)
receipt = tmp_path / 'recovery.json'
monkeypatch.setattr(sys, 'argv', ['isolated-authelia', '--transfer-receipt', str(transfer),
'--receipt', str(receipt), '--user', 'admin'])
assert m.main() == 0, receipt.read_text()
result = json.loads(receipt.read_text())
assert result['status'] == 'isolated_authelia_verified'
assert result['temporary_plaintext_cleanup_verified'] and result['container_cleanup_verified']
assert result['all_manifest_hashes_verified'] and result['issuer_key_matches_archive']
assert not result['production_contacted'] and not result['secret_values_exported']
text = capsys.readouterr().out + receipt.read_text()
assert PASSWORD not in text and CLIENT_SECRET not in text
assert not any(v in text for v in values.values())

View file

@ -29,6 +29,15 @@ grant. Actual-image synthetic tests and the real owner backup-recovery run
passed on 2026-10-16; T16 is done. See the packet and actual LLDAP receipt for
the exact target, resource limits, accepted scope and repeat-drill command.
`run-isolated-authelia.sh USER TRANSFER_RECEIPT` is the Authelia stage (T17,
prepared under T41). It asks for the same two private inputs. `isolated-authelia.py`
runs restored LLDAP as the network-none anchor with Authelia joining its
namespace, and checks the original config, storage encryption key, issuer
key, KeyCape's `keycape` client secret and existing-user first factor, with
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 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
the linked dump into a loopback-only PostgreSQL/container namespace, reads

View file

@ -0,0 +1,425 @@
#!/usr/bin/env python3
"""Owner-run native Authelia restore against the restored LLDAP directory.
Private inputs remain in the owner process/pipe. Decrypted staging uses Linux
tmpfs. Restored LLDAP is the network-none anchor; Authelia and the probe join
its namespace and use loopback only. No ports, host mounts or production APIs.
"""
import argparse
import base64
from datetime import datetime, timezone
import hashlib
import importlib.util
import json
import os
from pathlib import Path, PurePosixPath
import re
import signal
import subprocess
import tempfile
import time
from urllib.parse import urlsplit
import uuid
import yaml
HERE = Path(__file__).resolve().parent
def load(name):
spec = importlib.util.spec_from_file_location(name.replace('-', '_'), HERE / (name + '.py'))
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
# Reused unedited: isolated-lldap.py is in the pinned KeyCape attempt sources.
lldap = load('isolated-lldap')
verify, offline, bundle = lldap.verify, lldap.offline, lldap.bundle
command, archive_files = lldap.command, lldap.archive_files
IMAGE = 'docker.io/authelia/authelia@sha256:bd97cff4fcbf715b5ff1f9ae286afbe6033afce385302520b0368122d43a6f54'
LDAP_URL = 'ldap://127.0.0.1:3890'
NOTIFICATIONS = 'notification.txt'
def preflight():
target = lldap.preflight()
image_id = command(['docker', 'image', 'inspect', '--format', '{{.Id}}', IMAGE]).decode().strip()
target['images'].append({'reference': IMAGE, 'image_id': image_id})
return target
def mounted(spec, volumes, predicate):
found = [m for m in spec.get('volumeMounts', []) if predicate(volumes.get(m['name'], {}))]
if len(found) != 1:
raise bundle.Refused()
return found[0]
def configuration(root):
"""Assemble the archived Authelia runtime with only confinement rewrites."""
deployment = bundle.resource(root, 'sso', 'deployment', 'authelia')
pod = deployment['spec']['template']['spec']
main = next(c for c in pod['containers'] if c['name'] == 'authelia')
volumes = {v['name']: v for v in pod.get('volumes', [])}
secrets_mount = mounted(main, volumes, lambda v: v.get('secret', {}).get('secretName') == 'authelia-secrets')
config_mount = mounted(main, volumes, lambda v: v.get('configMap', {}).get('name') == 'authelia-config')
data_mount = mounted(main, volumes, lambda v: 'persistentVolumeClaim' in v)
smtp_paths = [m['mountPath'] for m in main.get('volumeMounts', [])
if volumes.get(m['name'], {}).get('secret', {}).get('secretName') == 'authelia-smtp']
secret_root = PurePosixPath(secrets_mount['mountPath'])
data_root = PurePosixPath(data_mount['mountPath'])
if not secret_root.is_absolute() or not data_root.is_absolute() or volumes[secrets_mount['name']]['secret'].get('items'):
raise bundle.Refused()
raw = bundle.resource(root, 'sso', 'secret', 'authelia-secrets')['data']
secrets = {}
for key, value in raw.items():
if not re.fullmatch(r'[A-Za-z0-9_.-]{1,128}', key):
raise bundle.Refused()
decoded = base64.b64decode(value, validate=True)
if not decoded:
raise bundle.Refused()
secrets[str(secret_root / key)] = decoded
env = {}
for entry in main.get('env', []):
if 'value' not in entry:
raise bundle.Refused()
name, value = entry['name'], entry['value']
if any(value.startswith(p.rstrip('/') + '/') for p in smtp_paths):
continue # Notifier credentials are never imported.
if name.endswith('_FILE') and value not in secrets:
raise bundle.Refused()
if not re.fullmatch(r'[A-Z][A-Z0-9_]{0,127}', name) or '\n' in value:
raise bundle.Refused()
env[name] = value
issuer_path = env.get('AUTHELIA_IDENTITY_PROVIDERS_OIDC_ISSUER_PRIVATE_KEY_FILE')
if not issuer_path or 'AUTHELIA_STORAGE_ENCRYPTION_KEY_FILE' not in env:
raise bundle.Refused()
configmap = bundle.resource(root, 'sso', 'configmap', 'authelia-config')['data']
config = yaml.safe_load(configmap[config_mount.get('subPath', 'configuration.yml')])
ldap = config['authentication_backend']['ldap']
if ('address' in ldap) == ('url' in ldap):
raise bundle.Refused()
ldap['address' if 'address' in ldap else 'url'] = LDAP_URL
config['notifier'] = {'filesystem': {'filename': str(data_root / NOTIFICATIONS)}}
# The NTP startup check is an outbound UDP call; it cannot succeed without a network.
config['ntp'] = {**(config.get('ntp') or {}), 'disable_startup_check': True}
store = PurePosixPath(config['storage']['local']['path'])
if store.parent != data_root or not (root / 'data/authelia' / store.name).is_file():
raise bundle.Refused()
server = config.setdefault('server', {})
# The healthcheck helper writes /app/.healthcheck.env before any provider check;
# the root filesystem stays read-only and the probe uses /api/health instead.
server['disable_healthcheck'] = True
if 'address' in server:
port = urlsplit(server['address']).port
else:
port = server.get('port', 9091)
if not isinstance(port, int) or not 0 < port < 65536:
raise bundle.Refused()
clients = config['identity_providers']['oidc']['clients']
keycape = yaml.safe_load(base64.b64decode(
bundle.resource(root, 'sso', 'secret', 'keycape-config')['data']['config.yaml'], validate=True))['authelia']
client = [c for c in clients if c.get('client_id', c.get('id')) == keycape['clientId']]
secret = keycape.get('clientSecret')
portal = urlsplit(keycape.get('browserBaseURL', ''))
if (len(client) != 1 or not isinstance(secret, str) or not secret or portal.scheme != 'https'
or not portal.hostname or not client[0].get('redirect_uris')):
raise bundle.Refused()
return {'config': yaml.safe_dump(config, sort_keys=False).encode(), 'config_path': config_mount['mountPath'],
'secrets': secrets, 'env': env, 'data_root': str(data_root), 'secret_root': str(secret_root),
'port': port, 'host': portal.hostname, 'issuer_key': secrets[issuer_path],
'client_id': keycape['clientId'], 'client_secret': secret,
'redirect_uri': client[0]['redirect_uris'][0]}
def public_jwk(pem):
from cryptography.hazmat.primitives import serialization
key = serialization.load_pem_private_key(pem, password=None).public_key()
numbers = key.public_numbers()
def encode(value):
raw = value.to_bytes((value.bit_length() + 7) // 8, 'big')
return base64.urlsafe_b64encode(raw).rstrip(b'=').decode()
jwk = {'e': encode(numbers.e), 'kty': 'RSA', 'n': encode(numbers.n)}
thumbprint = base64.urlsafe_b64encode(hashlib.sha256(
json.dumps(jwk, separators=(',', ':'), sort_keys=True).encode()).digest()).rstrip(b'=').decode()
return jwk, thumbprint
PROBE = r'''
import base64,json,sys,time,urllib.error,urllib.parse,urllib.request
from pathlib import Path
result={'status':'failed'}
stage='network_isolation'
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self,*args,**kwargs):return None
opener=urllib.request.build_opener(NoRedirect,urllib.request.ProxyHandler({}))
try:
assert {p.name for p in Path('/sys/class/net').iterdir()}=={'lo'}
assert len(Path('/proc/net/route').read_text().splitlines())==1
v=json.load(sys.stdin)
base='http://127.0.0.1:%d'%v['port']
head={'X-Forwarded-Proto':'https','X-Forwarded-Host':v['host'],'X-Forwarded-For':'127.0.0.1'}
def call(path,data=None,headers=None,form=False):
h=dict(head,**(headers or {}))
if data is not None:
if form:body=urllib.parse.urlencode(data).encode();h['Content-Type']='application/x-www-form-urlencoded'
else:body=json.dumps(data).encode();h['Content-Type']='application/json'
else:body=None
try:
with opener.open(urllib.request.Request(base+path,data=body,headers=h),timeout=10) as r:return r.status,r.read()
except urllib.error.HTTPError as e:return e.code,e.read()
stage='authelia_startup'
for _ in range(120):
try:
if call('/api/health')[0]==200:break
except OSError:pass
time.sleep(.25)
else:raise RuntimeError()
stage='issuer_jwks'
code,body=call('/jwks.json')
assert code==200
keys=[{'n':k['n'],'e':k['e']} for k in json.loads(body)['keys'] if k.get('kty')=='RSA' and k.get('use','sig')=='sig']
stage='existing_user_first_factor'
code,body=call('/api/firstfactor',{'username':v['user'],'password':v['password'],'keepMeLoggedIn':False})
assert code==200 and json.loads(body).get('status')=='OK'
stage='wrong_password_first_factor'
code,body=call('/api/firstfactor',{'username':v['user'],'password':v['wrong_password'],'keepMeLoggedIn':False})
assert code==401
def token(secret):
auth=base64.b64encode((urllib.parse.quote(v['client_id'],safe='')+':'+urllib.parse.quote(secret,safe='')).encode()).decode()
code,body=call('/api/oidc/token',{'grant_type':'authorization_code','code':'nk49-invalid-'+v['nonce'],
'redirect_uri':v['redirect_uri']},{'Authorization':'Basic '+auth},form=True)
return code,json.loads(body)
stage='client_secret_verifier'
code,body=token(v['client_secret'])
assert code==400 and body.get('error')=='invalid_grant' and 'access_token' not in body
stage='wrong_client_secret'
code,body=token(v['wrong_secret'])
assert code==401 and body.get('error')=='invalid_client' and 'access_token' not in body
result={'status':'verified','jwks_public_keys':keys,'existing_user_first_factor':True,
'wrong_password_refused':True,'archived_client_secret_accepted':True,
'wrong_client_secret_refused':True,'grant_issued':False,
'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))
'''
def contained(name, network, user='1000:1000'):
guard = json.loads(command(['docker', 'container', 'inspect', name, '--format',
'{"host":{{json .HostConfig}},"mounts":{{json .Mounts}},"user":{{json .Config.User}}}']))
host = guard['host']
if (host['NetworkMode'] != network or not host['ReadonlyRootfs'] or host['Privileged']
or host.get('Binds') or host.get('PortBindings') or 'ALL' not in host.get('CapDrop', [])
or 'no-new-privileges' not in host.get('SecurityOpt', []) or guard['user'] != user
or any(mount['Type'] != 'tmpfs' for mount in guard['mounts'])):
raise bundle.Refused()
def tmpfs(path, size):
return ['--tmpfs', path + ':rw,nosuid,nodev,noexec,size=' + str(size) + ',uid=1000,gid=1000,mode=0700']
def stopped(name, pidfile):
command(['docker', 'exec', name, '/bin/sh', '-c', 'kill -TERM "$(cat ' + pidfile + ')"'])
for _ in range(100):
running = subprocess.run(['docker', 'exec', name, '/bin/sh', '-c', 'kill -0 "$(cat ' + pidfile + ')"'],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10)
if running.returncode:
return
time.sleep(.1)
raise bundle.Refused()
def restore_authelia(root, user, password, receipt):
lldap_config, jwt, admin_password, _ = lldap.configuration(root)
runtime = configuration(root)
expected_jwk, thumbprint = public_jwk(runtime['issuer_key'])
stem = 'nk49-authelia-' + uuid.uuid4().hex
anchor, service, client = stem + '-lldap', stem + '-service', stem + '-client'
stage = 'directory_launch'
try:
command(['docker', 'run', '--detach', '--pull=never', '--name', anchor,
'--network', 'none', '--read-only', '--cap-drop', 'ALL',
'--security-opt', 'no-new-privileges', '--user', '1000:1000',
'--memory', '256m', '--cpus', '1', '--pids-limit', '128',
*tmpfs('/data', 268435456), *tmpfs('/run/recovery', 1048576), *tmpfs('/tmp', 16777216),
'--env', 'LLDAP_JWT_SECRET_FILE=/run/recovery/jwt',
'--env', 'LLDAP_LDAP_USER_PASS_FILE=/run/recovery/admin-password',
'--entrypoint', '/bin/sh', lldap.IMAGE, '-c', 'sleep 1800'])
contained(anchor, 'none')
anchor_id = command(['docker', 'container', 'inspect', anchor, '--format', '{{.Id}}']).decode().strip()
stage = 'authelia_launch'
# Environment carries file paths only; values arrive by tar into tmpfs.
args = ['docker', 'run', '--detach', '--pull=never', '--name', service,
'--network', 'container:' + anchor_id, '--read-only', '--cap-drop', 'ALL',
'--security-opt', 'no-new-privileges', '--user', '1000:1000',
'--memory', '512m', '--cpus', '1', '--pids-limit', '128',
*tmpfs(runtime['secret_root'], 1048576), *tmpfs(runtime['data_root'], 268435456),
*tmpfs(str(PurePosixPath(runtime['config_path']).parent), 1048576), *tmpfs('/tmp', 16777216)]
for name, value in runtime['env'].items():
args += ['--env', name + '=' + value]
command(args + ['--entrypoint', '/bin/sh', IMAGE, '-c', 'sleep 1800'])
contained(service, 'container:' + anchor_id)
receipt['container_isolation_verified_before_private_import'] = True
stage = 'private_import'
payload = {p.relative_to(root / 'data/lldap').as_posix(): p.read_bytes()
for p in (root / 'data/lldap').rglob('*') if p.is_file()}
command(['docker', 'exec', '-i', anchor, 'tar', '-xf', '-', '-C', '/data'], input=archive_files(payload))
command(['docker', 'exec', '-i', anchor, 'tar', '-xf', '-', '-C', '/run/recovery'],
input=archive_files({'lldap.toml': lldap_config, 'jwt': jwt, 'admin-password': admin_password}))
stored = {p.relative_to(root / 'data/authelia').as_posix(): p.read_bytes()
for p in (root / 'data/authelia').rglob('*') if p.is_file()}
command(['docker', 'exec', '-i', service, 'tar', '-xf', '-', '-C', runtime['data_root']],
input=archive_files(stored))
for directory, files in ((runtime['secret_root'], {k: v for k, v in runtime['secrets'].items()}),
(str(PurePosixPath(runtime['config_path']).parent),
{runtime['config_path']: runtime['config']})):
relative = {str(PurePosixPath(k).relative_to(directory)): v for k, v in files.items()}
command(['docker', 'exec', '-i', service, 'tar', '-xf', '-', '-C', directory], input=archive_files(relative))
stage = 'configuration_validation'
config = ['--config', runtime['config_path']]
command(['docker', 'exec', service, '/app/authelia', 'validate-config', *config])
receipt['original_configuration_validated'] = True
stage = 'storage_decryption'
checked = command(['docker', 'exec', service, '/app/authelia', 'storage', 'encryption', 'check', *config])
if b'SUCCESS' not in checked or b'FAILURE' in checked:
raise bundle.Refused()
receipt['storage_encryption_key_verified'] = True
stage = 'directory_startup'
command(['docker', 'exec', '--detach', anchor, '/bin/sh', '-c',
'echo $$ > /run/recovery/server.pid; exec /app/lldap run --config-file /run/recovery/lldap.toml > /dev/null 2>&1'])
# Authelia's startup check binds to LDAP once and exits if it is not up yet.
command(['docker', 'exec', service, '/bin/sh', '-c',
'i=0; until nc -w 1 127.0.0.1 3890 </dev/null >/dev/null 2>&1; do '
'i=$((i+1)); [ "$i" -ge 120 ] && exit 1; sleep 0.25; done'], timeout=60)
stage = 'service_startup'
command(['docker', 'exec', '--detach', service, '/bin/sh', '-c',
'echo $$ > /tmp/authelia.pid; exec /app/authelia ' + ' '.join(config) + ' > /dev/null 2>&1'])
stage = 'native_probe'
raw = command(['docker', 'run', '--interactive', '--rm', '--name', client, '--pull=never',
'--network', 'container:' + anchor_id, '--read-only', '--cap-drop', 'ALL',
'--security-opt', 'no-new-privileges', '--memory', '256m', '--cpus', '1',
'--pids-limit', '64', '--entrypoint', 'python3', lldap.CLIENT, '-c', PROBE],
input=json.dumps({'user': user, 'password': password, 'port': runtime['port'],
'host': runtime['host'], 'client_id': runtime['client_id'],
'client_secret': runtime['client_secret'],
'redirect_uri': runtime['redirect_uri'], 'nonce': uuid.uuid4().hex,
'wrong_password': 'nk49-wrong-' + uuid.uuid4().hex,
'wrong_secret': 'nk49-wrong-' + uuid.uuid4().hex}).encode(), timeout=180)
result = json.loads(raw)
keys = result.pop('jwks_public_keys', [])
receipt.update(result)
if result.get('status') != 'verified':
stage = result.get('failed_stage', stage)
raise bundle.Refused()
if {'n': expected_jwk['n'], 'e': expected_jwk['e']} not in keys:
stage = 'issuer_jwks'
raise bundle.Refused()
receipt.update(issuer_key_matches_archive=True, issuer_jwk_thumbprint_sha256=thumbprint)
stage = 'service_shutdown'
stopped(service, '/tmp/authelia.pid')
stopped(anchor, '/run/recovery/server.pid')
sent = subprocess.run(['docker', 'exec', service, 'test', '-s', runtime['data_root'] + '/' + NOTIFICATIONS],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10).returncode == 0
receipt.update(notifications_confined_to_tmpfs_file=True, notification_written=sent)
except Exception:
receipt['status'] = 'failed'
receipt.setdefault('failed_stage', stage)
raise
finally:
for owned in (client, service, anchor):
found = subprocess.run(['docker', 'container', 'inspect', owned], stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, timeout=15)
if found.returncode == 0:
command(['docker', 'rm', '--force', '--volumes', owned])
absent = subprocess.run(['docker', 'container', 'inspect', owned], stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, timeout=15)
command(['docker', 'info', '--format', '{{.ID}}'])
if absent.returncode == 0:
raise bundle.Refused()
receipt['container_cleanup_verified'] = True
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--transfer-receipt', required=True)
parser.add_argument('--receipt', required=True)
parser.add_argument('--user', required=True)
args = parser.parse_args()
os.umask(0o077)
fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
result = {'schema': 'netkingdom.isolated-authelia-recovery.v1', 'status': 'failed',
'production_contacted': False, 'secret_values_exported': False,
'container_cleanup_verified': False, 'temporary_plaintext_cleanup_verified': False,
'source_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
'reused_source_sha256': {name: hashlib.sha256((HERE / name).read_bytes()).hexdigest()
for name in ('isolated-lldap.py', 'identity-bundle.py', 'verify-identity.py',
'retrieve-identity-offline.py')}}
keyfd = None
work = None
stage = 'target_preflight'
try:
result['target'] = preflight()
stage = 'input_validation'
if not re.fullmatch(r'[A-Za-z0-9_.-]{1,128}', args.user):
raise bundle.Refused()
transfer = json.loads(Path(args.transfer_receipt).read_text())
offline.validate(transfer)
if transfer.get('independent_owner_retrieval_verified') is not True:
raise bundle.Refused()
path = Path(transfer['fetched_path'])
bundle.private(path, max_bytes=bundle.LIMIT)
if path.stat().st_size != transfer['ciphertext_bytes'] or offline.offsite.digest(path) != transfer['ciphertext_sha256']:
raise bundle.Refused()
keyfd = verify.custody.private_key_fd()
derived = command(['age-keygen', '-y', f'/proc/self/fd/{keyfd}'], pass_fds=(keyfd,)).strip()
if derived != bundle.profile()['public_recipient'].encode():
raise bundle.Refused()
stage = 'private_existing_password_prompt'
password = offline.private_prompt('Existing account password for the isolated Authelia check: ')
with tempfile.TemporaryDirectory(prefix='nk49-service-recovery-', dir='/dev/shm') as temporary:
work = Path(temporary)
archive = work / 'archive.tar'
stage = 'archive_decryption'
with archive.open('xb') as output:
proc = subprocess.run(['age', '-d', '-i', f'/proc/self/fd/{keyfd}', str(path)],
stdout=output, stderr=subprocess.DEVNULL, pass_fds=(keyfd,), timeout=180)
if proc.returncode or archive.stat().st_size > bundle.LIMIT + 4 * 1024 * 1024:
raise bundle.Refused()
restored = work / 'restored'
restored.mkdir(mode=0o700)
stage = 'archive_validation'
manifest, _ = bundle.restore(archive, restored)
result.update(ciphertext_sha256=transfer['ciphertext_sha256'],
manifest_file_count=len(manifest['files']), all_manifest_hashes_verified=True)
stage = 'native_authelia_check'
restore_authelia(restored, args.user, password, result)
password = None
if work.exists():
raise bundle.Refused()
result.update(status='isolated_authelia_verified', temporary_plaintext_cleanup_verified=True)
except (Exception, KeyboardInterrupt):
result.update(status='failed', error='isolated_authelia_check_failed_private_diagnostics_suppressed')
result.setdefault('failed_stage', stage)
finally:
if keyfd is not None:
os.close(keyfd)
if work is not None and not work.exists():
result['temporary_plaintext_cleanup_verified'] = True
result['finished_at'] = datetime.now(timezone.utc).isoformat()
with os.fdopen(fd, 'w') as output:
json.dump(result, output, indent=2)
output.write('\n')
success = result['status'] == 'isolated_authelia_verified'
print('OK: restored Authelia config, storage key, issuer key, client verifier, first factor and cleanup passed.'
if success else 'ERROR: isolated Authelia recovery refused; review the safe receipt. No production change.')
return 0 if success else 1
if __name__ == '__main__':
for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
signal.signal(sig, offline.interrupted)
raise SystemExit(main())

View file

@ -0,0 +1,37 @@
#!/usr/bin/env bash
# Owner-run existing-user check. No private key/password belongs in arguments.
set +x
set -euo pipefail
ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/../.." && pwd)"
cd -- "$ROOT"
umask 077
[[ $# == 2 ]] || { printf '%s\n' 'Usage: run-isolated-authelia.sh USER VERIFIED_TRANSFER_RECEIPT'; exit 1; }
[[ "$1" =~ ^[A-Za-z0-9_.-]{1,128}$ ]] || { printf '%s\n' 'ERROR: invalid username.'; exit 1; }
[[ -f "$2" && ! -L "$2" ]] || { printf '%s\n' 'ERROR: provide the existing verified transfer receipt.'; exit 1; }
[[ -t 0 && -t 1 && -t 2 ]] || { printf '%s\n' 'ERROR: run in your own interactive terminal.'; exit 1; }
for tool in python3 age age-keygen openssl docker; do
command -v "$tool" >/dev/null || { printf 'ERROR: required tool unavailable: %s\n' "$tool"; exit 1; }
done
python3 - <<'PY'
import importlib.util
from pathlib import Path
p=Path('tools/bootstrap-recovery/isolated-authelia.py')
spec=importlib.util.spec_from_file_location('isolated',p)
m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
try:
m.preflight()
except Exception:
raise SystemExit('ERROR: exact local Docker target, pinned images or tmpfs preflight failed; no key requested.')
print('OK: exact bnt-lap001 recovery engine, pinned images and tmpfs available.')
PY
NK49_AUTHELIA_RUN="$(mktemp -d /tmp/nk49-isolated-authelia-XXXXXXXX)"
printf 'Safe evidence directory: %s\n' "$NK49_AUTHELIA_RUN"
printf '%s\n' 'This restores LLDAP and Authelia into disposable containers sharing one network-free namespace: no external network, ports or host mounts.'
printf '%s\n' 'Enter the railiance-backup age key, then the existing account password that was valid at backup capture.'
printf '%s\n' 'Checks: original config, storage key, issuer key, KeyCape client verifier, first factor; wrong password and client secret refused. Notifications go to a tmpfs file. No production change.'
bash sso-mfa/bootstrap/sops-custody-unlock.sh \
--expected-recipient age1zvryunvjhvpkmasskauga2heeg0ztnte9ymgppvjge36ekumk50syr3tsz \
--prompt -- python3 tools/bootstrap-recovery/isolated-authelia.py \
--user "$1" --transfer-receipt "$2" --receipt "$NK49_AUTHELIA_RUN/recovery.json"
printf 'Safe receipt: %s/recovery.json\n' "$NK49_AUTHELIA_RUN"
printf '%s\n' 'Return only the outcome and evidence directory. KeyCape consumer, MFA and audit acceptance remain separate.'

View file

@ -509,7 +509,7 @@ status: wait
priority: high
assignee: codex
depends_on: [NK-WP-0049-T15, NK-WP-0049-T16, NK-WP-0049-T41]
blocking_reason: "LLDAP recovery T16 passed. Await the T41 adapter (prepared and fixture-tested), then a founder-attended run with the age key and an existing-user password."
blocking_reason: "LLDAP T16 passed; T41 stage prepared and fixture-tested 2026-10-24. Await the founder-attended run: run-isolated-authelia.sh with a fresh verified transfer receipt, the age key and an existing-user password."
state_hub_task_id: "b47f9975-d6d7-5bb8-93ce-886deb1dfa70"
```
@ -1471,7 +1471,7 @@ OpenBao login, rotation, production action or owner assent was exercised.
```task
id: NK-WP-0049-T41
status: progress
status: done
priority: high
assignee: claude-code
depends_on: [NK-WP-0049-T16, NK-WP-0049-T29]
@ -1508,6 +1508,19 @@ local images with disposable credentials only. The packet in
`docs/identity-service-restore-packet.md` names the stage and its exact command.
This prepares T17; it is not T17 acceptance and touches no production service.
Completed 2026-10-24. `isolated-authelia.py`, `run-isolated-authelia.sh` and
`tests/test_isolated_authelia_recovery.py` are in place. Unit tests and the
opt-in native fixture tests pass on the pinned local images (9 passed): positive
path, wrong password at `existing_user_first_factor`, wrong storage key at
`storage_decryption`, and an encrypted-bundle run through `main()`. Native runs
surfaced two confinement rewrites beyond the planned ones: the NTP startup
check (outbound UDP) and the container healthcheck file (Authelia writes
`/app/.healthcheck.env` before provider checks, which a read-only root refuses).
Both are recorded in the packet. `isolated-lldap.py` and the KeyCape attempt
sources are unedited; `isolated-keycape.py` still hashes to the T38 pin
`86f37add`. The packet section "Prepared owner step: isolated Authelia
recovery (T17)" gives the exact command.
## Accept remaining recovery and caller evidence and close the G08 handoff
```task