railiance-platform/scripts/root_custody_finish.py

#!/usr/bin/env python3
"""Silent attended current-source and governed provider history recovery proof."""
import argparse
import configparser
import hmac
import re
from urllib.parse import urlsplit,unquote
from datetime import datetime,timezone
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import tempfile
import uuid

from root_grant_attended import Bao, RECORD_ID, atomic_json, check_baseline, digest, require_attended
from migrate_nextcloud_backup_account import HOST, OPERATOR_PATH, LANE, FOLDER, LABEL, quota, request, ocs

VALIDATOR_DIGEST='9156e399c8d92a45b817bfc3738c89843487ce57e1718ed747f804f95cc719bb'
SCHEMA_DIGEST='bb6cd6da5dc48eddcf118ae78c2cacfafd5f860098fd8a8368fe6bc845776d07'
INDEPENDENT_ESCROW=Path('/home/worsch/.config/age/railiance-backup.key')


def checked_bundle(state_path, validator):
    state=json.loads(state_path.read_text())
    if state.get('pending') or validator.validate(state['document']) or state['record_id']!=RECORD_ID or digest(state['document'])!=state['digest']:
        raise FinishFailure('retained_state_invalid')
    history=[]
    for path in sorted((state_path.parent/'history').glob('*.json')):
        if path.is_symlink() or path.stat().st_size>262144:
            raise FinishFailure('history_invalid')
        receipt=json.loads(path.read_text())
        if validator.validate(receipt['document']) or receipt['record_id']!=RECORD_ID or receipt['digest']!=digest(receipt['document']):
            raise FinishFailure('history_invalid')
        history.append(receipt)
    if state not in history:
        raise FinishFailure('latest_history_missing')
    return state,json.dumps({'baseline':state,'history':history},sort_keys=True,separators=(',',':')).encode()


def historical_config_comparison(config_path, values):
    if not config_path.exists():
        return {"historical_config_present":False,"historical_share_matches_current_lane":None}
    try:
        config=configparser.ConfigParser(interpolation=None)
        config.read(config_path)
        remote=config['nextcloud-backup']
        candidates=[remote.get('user','')]+unquote(urlsplit(remote.get('url','')).path).split('/')
        matched=any(hmac.compare_digest(value,values['NC_WEBDAV_TOKEN']) for value in candidates if value)
        return {"historical_config_present":True,"historical_share_matches_current_lane":matched,
                "historical_credential_status":"current-lane-incident" if matched else "historical-owner-and-revocation-unverified"}
    except Exception:
        return {"historical_config_present":True,"historical_share_matches_current_lane":None,
                "historical_credential_status":"comparison-unavailable"}



def inspect_share(lane, operator, inventory=None, network=None, progress=lambda stage, **details: None):
    """Owner metadata and depth-zero public probes only; never upload or change shares."""
    inventory=ocs if inventory is None else inventory
    network=request if network is None else network
    values=lane['data']
    if operator['BACKUP_USERNAME']!='Backup':
        raise FinishFailure('governed_owner_mismatch')
    token=values['NC_WEBDAV_TOKEN']
    upload=values['NC_WEBDAV_URL']
    if not token or upload != HOST+'/public.php/dav/filesdrop/'+token:
        raise FinishFailure('current_lane_contract_mismatch')
    auth=(operator['BACKUP_USERNAME'],operator['BACKUP_PASSWORD'])
    progress('share_inventory')
    # Existing owner-scoped inventory, filtered provider-side to reviewed folder.
    from urllib.parse import quote
    shares=inventory(auth,'?path='+quote(FOLDER,safe=''))
    if not isinstance(shares,list) or len(shares)>1000:
        raise FinishFailure('share_inventory_invalid')
    labelled=[x for x in shares if x.get('label')==LABEL]
    bound=[x for x in shares if isinstance(x.get('token'),str) and hmac.compare_digest(x['token'],token)]
    result={'labelled_share_present':bool(labelled),'labelled_share_unique':len(labelled)==1,
            'current_token_found_in_reviewed_folder':bool(bound),'current_token_unique':len(bound)==1,
            'read_only_diagnostic':True,'provider_mutations_performed':False,'backup_roundtrip_proved':False}
    if len(bound)==1:
        share=bound[0]
        expiry=share.get('expiration')
        expired=None
        if expiry:
            try:
                parsed=datetime.fromisoformat(expiry)
                if parsed.tzinfo is not None:
                    expired=parsed.astimezone(timezone.utc)<=datetime.now(timezone.utc)
            except (TypeError,ValueError):pass
        result.update(current_share_label_matches=share.get('label')==LABEL,
          current_share_owner_matches=share.get('uid_owner')=='Backup',
          current_share_path_matches=share.get('path')==FOLDER,
          current_share_public_link=share.get('share_type') in (3,'3'),
          current_share_create_only=share.get('permissions') in (4,'4'),
          current_share_password_indicator=bool(share['password']) if 'password' in share else None,
          current_share_password_indicator_known='password' in share,
          current_share_expiry_present=bool(expiry),current_share_expired=expired,
          current_share_expiry_clock_basis_known=not bool(expiry) or expired is not None)
        if result['current_share_password_indicator'] is True or expired is True:
            result['public_route_probes_skipped']=True
            result['public_route_skip_reason']='password_protected' if result['current_share_password_indicator'] is True else 'expired'
            return result
    result['public_route_probes_skipped']=False
    progress('share_route_probes')
    body=b'<d:propfind xmlns:d="DAV:"><d:prop><d:resourcetype/></d:prop></d:propfind>'
    headers={'Depth':'0','Content-Type':'application/xml','X-Requested-With':'XMLHttpRequest'}
    # No listing of children, no PUT; response bodies remain captured and discarded.
    for name,url,credentials in (
        ('current_route',upload,(token,'')),
        ('documented_route',HOST+'/public.php/dav/files/'+token,(token,'')),
        ('documented_route_no_basic',HOST+'/public.php/dav/files/'+token,None)):
        code,_=network(url,'PROPFIND',body,credentials,headers)
        if type(code) is not int or not 100<=code<=599:
            raise FinishFailure('share_probe_invalid')
        result[name+'_http_status']=code
        result[name+'_classification']=('authentication_rejected' if code==401 else
            'capability_denied' if code==403 else 'not_found' if code==404 else
            'properties_returned' if code==207 else 'other_status')
    return result


def repair_share(bundle, lane, operator, bao, expected_version, inventory=None, network=None, progress=lambda stage, **details: None):
    """Bounded owner cutover: proof first, CAS coordinates only, preserve all old shares."""
    inventory=ocs if inventory is None else inventory
    network=request if network is None else network
    if lane['metadata']['version']!=expected_version:
        raise FinishFailure('lane_version_changed')
    if operator['BACKUP_USERNAME']!='Backup':
        raise FinishFailure('governed_owner_mismatch')
    auth=(operator['BACKUP_USERNAME'],operator['BACKUP_PASSWORD'])
    progress('repair_inventory')
    # All current account-owned shares: detect label ambiguity outside exact folder too.
    shares=inventory(auth)
    if not isinstance(shares,list) or len(shares)>1000:
        raise FinishFailure('share_inventory_invalid')
    labelled=[x for x in shares if x.get('label')==LABEL]
    if len(labelled)>1:
        raise FinishFailure('repair_share_ambiguous')
    def verify(share):
        if not (share.get('uid_owner')=='Backup' and share.get('path')==FOLDER and
                share.get('label')==LABEL and share.get('permissions') in (4,'4') and
                share.get('share_type') in (3,'3') and 'password' in share and
                share['password'] in (None,'',False) and 'expiration' in share and
                share['expiration'] in (None,'') and isinstance(share.get('token'),str) and
                re.fullmatch(r'[A-Za-z0-9_-]+',share['token']) and isinstance(share.get('id'),(str,int)) and str(share['id']).isdigit()):
            raise FinishFailure('repair_share_scope_invalid')
    if labelled:
        share=labelled[0];verify(share)
        progress('repair_share_verified',replacement_share_created=False,replacement_share_id=int(share['id']))
    else:
        # Recheck the exact reviewed directory and valid Backup account before creation.
        get_quota=quota
        get_quota(auth)
        code,_=network(HOST+'/remote.php/dav/files/Backup'+FOLDER,'PROPFIND',
            b'<d:propfind xmlns:d="DAV:"><d:prop><d:resourcetype/></d:prop></d:propfind>',auth,
            {'Depth':'0','Content-Type':'application/xml'})
        if code!=207:
            raise FinishFailure('repair_folder_unverified')
        progress('repair_share_create')
        share=inventory(auth,method='POST',fields={'path':FOLDER,'shareType':3,'permissions':4,'label':LABEL,'sendMail':'false','expireDate':''})
        if isinstance(share.get('id'),(str,int)) and str(share['id']).isdigit():
            progress('repair_share_create',replacement_share_created=True,replacement_share_id=int(share['id']))
        verify(share)
    # Re-read the one admitted share: never trust just POST response.
    admitted=inventory(auth,'/'+str(share['id']))
    if not isinstance(admitted,list) or len(admitted)!=1:
        raise FinishFailure('repair_share_scope_invalid')
    admitted=admitted[0]
    verify(admitted)
    if not hmac.compare_digest(admitted['token'],share['token']):
        raise FinishFailure('repair_share_changed')
    progress('repair_share_verified',replacement_share_id=int(share['id']))
    updated=dict(lane['data'],NC_WEBDAV_TOKEN=share['token'],
                 NC_WEBDAV_URL=HOST+'/public.php/dav/files/'+share['token'])
    candidate={'data':updated,'metadata':lane['metadata']}
    proof=roundtrip(bundle,candidate,operator,network=network,progress=progress)
    # Archive remains stored on failure; no deletes of unknown shares/files.
    progress('lane_cas',provider_proof=proof)
    current=bao.read(LANE)
    if current!=lane:
        raise FinishFailure('lane_version_changed')
    if updated==lane['data']:
        progress('lane_cas',lane_cas_noop=True,published_lane_kv_version=expected_version)
        version=expected_version
    else:
        response=bao.write(LANE,{'options':{'cas':expected_version},'data':updated})
        version=response['data']['version']
        progress('lane_cas',lane_cas_noop=False,published_lane_kv_version=version)
        if version!=expected_version+1:
            raise FinishFailure('lane_cas_receipt_invalid')
    progress('lane_readback')
    readback=bao.read(LANE)
    if readback['metadata']['version']!=version or readback['data']!=updated:
        raise FinishFailure('lane_readback_mismatch')
    return dict(proof,published_lane_kv_version=version,coordinates_only_updated=True,
                escrow_and_unrelated_fields_preserved=True,unknown_shares_and_old_backups_preserved=True,
                runtime_consumers_refreshed=False)

def roundtrip(bundle, lane, operator, network=None, get_quota=None, progress=lambda stage, **details: None):
    network=request if network is None else network
    get_quota=quota if get_quota is None else get_quota
    progress('provider_contract')
    values=lane['data']
    if operator['BACKUP_USERNAME']!='Backup':
        raise FinishFailure('governed_owner_mismatch')
    auth=(operator['BACKUP_USERNAME'],operator['BACKUP_PASSWORD'])
    upload=values['NC_WEBDAV_URL']
    if upload not in (HOST+'/public.php/dav/filesdrop/'+values['NC_WEBDAV_TOKEN'],HOST+'/public.php/dav/files/'+values['NC_WEBDAV_TOKEN']):
        raise FinishFailure('current_lane_contract_mismatch')
    key=(values['AGE_PRIVATE_KEY'].strip()+'\n').encode()
    def process(command,payload):
        try:
            result=subprocess.run(command,input=payload,capture_output=True,timeout=30)
        except FileNotFoundError:
            raise FinishFailure('crypto_binary_unavailable') from None
        except subprocess.TimeoutExpired:
            raise FinishFailure('crypto_command_timeout') from None
        if result.returncode:
            raise FinishFailure('crypto_command_failed')
        return result.stdout
    progress('derive_age_recipient')
    recipient=process(['age-keygen','-y','/dev/stdin'],key).decode().strip()
    progress('encrypt_history')
    encrypted=process(['age','-r',recipient],bundle)
    progress('independent_key_escrow')
    if not INDEPENDENT_ESCROW.is_file():
        raise FinishFailure('independent_key_escrow_missing')
    independent_recipient=process(['age-keygen','-y',str(INDEPENDENT_ESCROW)],b'').decode().strip()
    matched=hmac.compare_digest(recipient,independent_recipient)
    progress('independent_key_escrow',independent_escrow_recipient_matches_current_lane=matched)
    if not matched:
        raise FinishFailure('independent_key_escrow_mismatch')
    progress('provider_quota')
    if len(encrypted)>1048576 or get_quota(auth)['available_bytes']<len(encrypted):
        raise FinishFailure('backup_size_or_quota_invalid')
    name='root-custody-history-'+datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')+'-'+uuid.uuid4().hex+'.json.age'
    progress('provider_upload',artifact=name)
    status,_=network(upload+'/forgejo/'+name,'PUT',encrypted,(values['NC_WEBDAV_TOKEN'],''),{'Content-Type':'application/octet-stream','If-None-Match':'*','X-Requested-With':'XMLHttpRequest'})
    progress('provider_upload',upload_http_status=status)
    if status!=201:
        raise FinishFailure('new_artifact_upload_failed')
    progress('provider_download')
    status,fetched=network(HOST+'/remote.php/dav/files/Backup/railiance-backups/forgejo/'+name,'GET',auth=auth)
    progress('provider_download',readback_http_status=status)
    if status!=200:
        raise FinishFailure('provider_download_failed')
    progress('provider_ciphertext_check')
    if hashlib.sha256(fetched).digest()!=hashlib.sha256(encrypted).digest():
        raise FinishFailure('provider_ciphertext_mismatch')
    progress('decrypt_and_compare_history')
    with tempfile.TemporaryDirectory(prefix='root-custody-recovery-') as tmp:
        cipher=Path(tmp)/'history.json.age'
        cipher.write_bytes(fetched)
        cipher.chmod(0o600)
        restored=process(['age','-d','-i',str(INDEPENDENT_ESCROW),str(cipher)],b'')
    if restored!=bundle:
        raise FinishFailure('decrypted_history_mismatch')
    return {'provider':'existing-governed-nextcloud-backup','artifact':name,'upload_http_status':201,'readback_http_status':200,'ciphertext_sha256':hashlib.sha256(encrypted).hexdigest(),'decrypted_history_matches':True,'lane_kv_version':lane['metadata']['version'],'local_historical_rclone_used':False,'independent_escrow_recipient_matches_current_lane':True,'independent_escrow_decryption_verified':True}


STAGES = {
    'repair_inventory','repair_share_create','repair_share_verified','lane_cas','lane_readback', 'share_inventory','share_route_probes', 'startup', 'attended_envelope', 'operator_endpoint', 'canon_pins', 'validator_import',
    'publisher_lock', 'retained_history', 'current_source_read', 'current_baseline_check',
    'backup_lane_read', 'backup_operator_read', 'historical_share_comparison',
    'provider_contract', 'derive_age_recipient', 'encrypt_history', 'independent_key_escrow', 'provider_quota',
    'provider_upload', 'provider_download', 'provider_ciphertext_check', 'decrypt_and_compare_history', 'complete',
}


class FinishFailure(ValueError):
    def __init__(self, code):
        self.code = code


ERROR_CODES = {'lane_version_changed','repair_share_ambiguous','repair_share_scope_invalid','repair_folder_unverified','repair_share_changed','lane_cas_receipt_invalid','lane_readback_mismatch','repair_confirmation_required','share_inventory_invalid','share_probe_invalid','stage_failed', 'attended_envelope_required', 'approved_operator_endpoint_required',
               'canon_pin_mismatch', 'current_backup_lane_incident', 'unexpected_stage',
               'crypto_binary_unavailable','crypto_command_timeout','crypto_command_failed','independent_key_escrow_missing','independent_key_escrow_mismatch',
               'provider_download_failed','provider_ciphertext_mismatch','decrypted_history_mismatch',
               'retained_state_invalid','history_invalid','latest_history_missing',
               'governed_owner_mismatch','current_lane_contract_mismatch','backup_size_or_quota_invalid','new_artifact_upload_failed'}


def main(receipt_reserved=False):
    p=argparse.ArgumentParser()
    modes=p.add_mutually_exclusive_group()
    modes.add_argument('--inspect-share',action='store_true')
    modes.add_argument('--repair-share',action='store_true')
    p.add_argument('--expected-lane-version',type=int)
    p.add_argument('--confirm-repair')
    p.add_argument('--state',type=Path,required=True)
    p.add_argument('--net-kingdom',type=Path,required=True)
    p.add_argument('--receipt',type=Path,required=True)
    args=p.parse_args()
    result={'ok':False,'stage':'startup','status':'started',
            'recorded_at':datetime.now(timezone.utc).isoformat()}
    def progress(stage, **details):
        if stage not in STAGES:
            raise FinishFailure('unexpected_stage')
        result['stage']=stage
        for key,value in details.items():
            if key=='provider_proof' and isinstance(value,dict):
                for name in ('provider','artifact','upload_http_status','readback_http_status','ciphertext_sha256','decrypted_history_matches','independent_escrow_recipient_matches_current_lane','independent_escrow_decryption_verified'):
                    if name in value:result[name]=value[name]
                result['provider_roundtrip_verified_before_cas']=True
                continue
            if key in {'replacement_share_created','lane_cas_noop'} and type(value) is bool:
                result[key]=value
                continue
            if key in {'replacement_share_id','published_lane_kv_version'} and type(value) is int and value>0:
                result[key]=value
                continue
            if key in {'upload_http_status','readback_http_status'} and type(value) is int and 100 <= value <= 599:
                result[key]=value
            elif key=='independent_escrow_recipient_matches_current_lane' and type(value) is bool:
                result[key]=value
            elif key=='artifact' and isinstance(value,str) and value.startswith('root-custody-history-') and value.endswith('.json.age'):
                result[key]=value
        atomic_json(args.receipt,result)
    try:
        if args.receipt.is_symlink():
            return 1
        if not receipt_reserved:
            fd=os.open(args.receipt,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600)
            os.close(fd)
        progress('attended_envelope')
        try:
            require_attended()
        except Exception:
            raise FinishFailure('attended_envelope_required') from None
        progress('operator_endpoint')
        if os.getenv('BAO_ADDR')!='http://127.0.0.1:18200':
            raise FinishFailure('approved_operator_endpoint_required')
        progress('canon_pins')
        source=args.net_kingdom/'tools/custody-record/validate.py'
        schema=args.net_kingdom/'canon/schemas/custody-record_v0.1.schema.json'
        if hashlib.sha256(source.read_bytes()).hexdigest()!=VALIDATOR_DIGEST or hashlib.sha256(schema.read_bytes()).hexdigest()!=SCHEMA_DIGEST:
            raise FinishFailure('canon_pin_mismatch')
        progress('validator_import')
        spec=importlib.util.spec_from_file_location('custody_validator',source)
        validator=importlib.util.module_from_spec(spec);spec.loader.exec_module(validator)
        import fcntl
        progress('publisher_lock')
        with (args.state.parent/'publisher.lock').open('a') as lock:
            fcntl.flock(lock,fcntl.LOCK_EX)
            progress('retained_history')
            state,bundle=checked_bundle(args.state,validator)
            bao=Bao()
            progress('current_source_read')
            actual=bao.read('platform/data/custody/net-kingdom/platform-root')
            progress('current_baseline_check')
            check_baseline(actual,state,validator)
            result.update(record_id=RECORD_ID,revision=state['revision'],kv_version=state['kv_version'],
                          digest=state['digest'],current_live_baseline_matches=True)
            progress('backup_lane_read')
            lane=bao.read(LANE)
            result['lane_kv_version']=lane['metadata']['version']
            progress('backup_operator_read')
            operator=bao.read(OPERATOR_PATH)['data']
            progress('historical_share_comparison')
            comparison=historical_config_comparison(Path('/home/worsch/.config/rclone/rclone.conf'),lane['data'])
            result.update(comparison)
            if comparison['historical_share_matches_current_lane'] is True:
                raise FinishFailure('current_backup_lane_incident')
            if args.inspect_share:
                diagnostic=inspect_share(lane,operator,progress=progress)
                result.update(ok=True,status='diagnostic-complete',stage='complete',
                    scope='read-only-current-governed-share-inspection',
                    finished_at=datetime.now(timezone.utc).isoformat(),**diagnostic)
                atomic_json(args.receipt,result)
                return 0
            if args.repair_share:
                if args.confirm_repair!='RESTORE Backup CREATE ONLY' or args.expected_lane_version is None:
                    raise FinishFailure('repair_confirmation_required')
                receipt=repair_share(bundle,lane,operator,bao,args.expected_lane_version,progress=progress)
            else:
                receipt=roundtrip(bundle,lane,operator,progress=progress)
        result.update(ok=True,status='verified',stage='complete',finished_at=datetime.now(timezone.utc).isoformat(),
           scope='current-owner-read-and-governed-history-provider-roundtrip',
           history_entries=len(json.loads(bundle)['history']),**receipt)
        atomic_json(args.receipt,result)
        return 0
    except FileExistsError:
        # Never change an old success receipt or claim it belongs to this attempt.
        return 1
    except Exception as error:
        result.update(status='failed',finished_at=datetime.now(timezone.utc).isoformat(),error_code=error.code if isinstance(error,FinishFailure) and error.code in ERROR_CODES else 'stage_failed')
        try:
            atomic_json(args.receipt,result)
        except Exception:
            pass
        return 1


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