net-kingdom/tools/bootstrap-recovery/prepare-isolated-keycape.py

#!/usr/bin/env python3
"""Validate reviewed public recovery metadata before requesting custody input.

Never opens TLS material, archive plaintext or any key. This validates bindings,
not owner assent, archive contents or recovery authority.
"""
import argparse
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import re
import stat
from urllib.parse import urlsplit

ROOT=Path(__file__).resolve().parents[2]
# The executable closure extends the synthetic fixture closure with the attended
# custody entry point and this validator. Tests and historical receipts do not run.
SOURCE_PATHS=tuple(sorted([
 'sso-mfa/bootstrap/sops-custody-unlock.sh','sso-mfa/bootstrap/sops-private.py','sso-mfa/bootstrap/sops-custody-fd.py',
 'sso-mfa/k8s/privacyidea/key-custody-remote.py',
 *('tools/bootstrap-recovery/'+name for name in (
 'isolated-keycape.py','isolated-keycape-session.py','run-isolated-keycape.sh',
 'prepare-isolated-keycape.py','isolated-lldap.py','identity-bundle.py',
 'identity-backup-profile.json','retrieve-identity-offline.py','identity-offsite.py',
 'verify-identity.py','identity-custody.json','identity-files.py',
 'check-privacyidea-database.py'))]))

class Refused(Exception):pass

def require(value):
    if not value:raise Refused()

def metadata_path(path):
    path=Path(os.path.abspath(path))
    require(not any(p.is_symlink() for p in (path,*path.parents)))
    st=path.stat()
    require(stat.S_ISREG(st.st_mode) and st.st_uid==os.getuid() and st.st_nlink==1 and not st.st_mode&0o077)
    return path

def read_metadata(path,limit=128*1024):
    path=metadata_path(path)
    require(0<path.stat().st_size<=limit)
    fd=os.open(path,os.O_RDONLY|os.O_NOFOLLOW|os.O_NONBLOCK)
    with os.fdopen(fd,'rb') as source:
        info=os.fstat(source.fileno())
        require(stat.S_ISREG(info.st_mode) and info.st_uid==os.getuid() and info.st_nlink==1 and not info.st_mode&0o077 and 0<info.st_size<=limit)
        raw=source.read(limit+1)
    require(0<len(raw)<=limit)
    return raw


def strict_json(raw):
    def unique(pairs):
        result={}
        for key,value in pairs:
            require(key not in result)
            result[key]=value
        return result
    return json.loads(raw,object_pairs_hook=unique)


def validate(transfer,admission,root=ROOT):
    require(isinstance(admission,dict))
    for field in ('ciphertext_sha256','verifier_sha256','trust_sha256','tls_cert_sha256'):
        require(isinstance(admission.get(field),str) and re.fullmatch('[a-f0-9]{64}',admission[field]))
    require(admission['ciphertext_sha256']==transfer.get('ciphertext_sha256'))
    require(transfer.get('independent_owner_retrieval_verified') is True)
    u=urlsplit(admission.get('issuer',''))
    require(u.scheme=='https' and u.hostname and not u.username and not u.password and not u.query and not u.fragment and u.port in (None,443) and u.path in ('','/'))
    consumer=admission.get('consumer',{})
    fields=('client_id','secret_env','audience','scope','subject','tenant','roles','deny_scope')
    require(isinstance(consumer,dict) and set(consumer)==set(fields))
    for field in fields:
        value=consumer[field]
        require(isinstance(value,str) and 0<len(value)<=1024 and value.strip()==value and not any(ord(c)<32 or ord(c)==127 for c in value))
    require(re.fullmatch('KEYCAPE_[A-Z0-9_]+',consumer['secret_env']))
    require(not set(consumer['scope'].split())&set(consumer['deny_scope'].split()))
    sources=admission.get('source_sha256')
    require(isinstance(sources,dict) and set(sources)==set(SOURCE_PATHS))
    for name,expected in sources.items():
        require(isinstance(expected,str) and re.fullmatch('sha256:[a-f0-9]{64}',expected))
        path=root/name
        require(not any(p.is_symlink() for p in (path,*path.parents)))
        require('sha256:'+hashlib.sha256(path.read_bytes()).hexdigest()==expected)
    spec=importlib.util.spec_from_file_location('adapter',root/'tools/bootstrap-recovery/isolated-keycape.py')
    adapter=importlib.util.module_from_spec(spec);spec.loader.exec_module(adapter)
    adapter.manifest_binding_mode(admission,transfer)
    # Validate provider receipt through the existing fixed contract, with no I/O.
    spec=importlib.util.spec_from_file_location('offline',root/'tools/bootstrap-recovery/retrieve-identity-offline.py')
    offline=importlib.util.module_from_spec(spec);spec.loader.exec_module(offline)
    offline.validate(transfer)
    ciphertext=metadata_path(transfer['fetched_path']) # no content read
    require(ciphertext.stat().st_size==transfer['ciphertext_bytes'])
    return True

def main():
    p=argparse.ArgumentParser(description=__doc__)
    for name in ('transfer-receipt','admission','admission-sha256','tls-cert','tls-key','trust'):p.add_argument('--'+name,required=True)
    args=p.parse_args()
    try:
        require(re.fullmatch('[a-f0-9]{64}',args.admission_sha256))
        raw=read_metadata(args.admission)
        require(hashlib.sha256(raw).hexdigest()==args.admission_sha256)
        transfer_raw=read_metadata(args.transfer_receipt)
        validate(strict_json(transfer_raw),strict_json(raw))
        for path in (args.tls_cert,args.tls_key,args.trust):metadata_path(path)
        # Import all third-party dependencies before the attended custody prompt.
        import yaml
        from cryptography import x509
        from cryptography.hazmat.primitives import serialization
    except Exception:
        print('ERROR: reviewed metadata/source/dependency preflight refused; no key requested.')
        return 1
    print('OK: reviewed metadata and source closure verified; private contents not inspected.')
    return 0

if __name__=='__main__':raise SystemExit(main())