railiance-platform/scripts/flex_auth_signer_custody.py

#!/usr/bin/env python3
"""Protected signer CAS primitive. Disposable owner fixture only; no live entry.

This module deliberately cannot admit production custody. A reviewed native CCR,
protected executor, fencing and founder scope are prerequisites for that adapter.
"""
from __future__ import annotations
from dataclasses import dataclass,field
from datetime import datetime,timezone
import base64
import hashlib
import hmac
import http.client
import json
import os
from pathlib import Path
import re
import secrets
import subprocess

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from task_execution_journal import Journal,Refused
from task_execution_native_fixture import _absolute_budget
from task_execution_recovery import _json_unique,_invalid_constant

PATH='platform/data/workloads/flex-auth/envelope-signing'
FIELDS={'ED25519_PRIVATE_KEY','ED25519_PUBLIC_KEY','kid'}
BINDINGS={'proposal_sha256','executed_revision_sha256','deployment_sha256','image_sha256','policy_sha256','trust_sha256'}


def encoded(value):return json.dumps(value,sort_keys=True,separators=(',',':'),allow_nan=False).encode()
def b64(value):return base64.urlsafe_b64encode(value).decode().rstrip('=')


@dataclass(frozen=True)
class FixtureWriterAdmission:
    port:int
    owner_token:str=field(repr=False)
    server_process:subprocess.Popen=field(repr=False)
    fenced:bool=False


class SignerWriter:
    """Single attempted operation; failed/unknown writes never retry or remint.

    File permissions are accidental-disclosure guards, not same-UID isolation.
    The receipt directory must be independent owner custody in a native adapter.
    Python cannot promise seed zeroization; process lifetime is the boundary.
    """
    def __init__(self,admission:FixtureWriterAdmission,directory:Path,operation_id:str,*,kid:str,expected_version:int,bindings:dict):
        if (type(admission) is not FixtureWriterAdmission or type(admission.port) is not int
                or not 1<=admission.port<=65535 or type(admission.owner_token) is not str
                or not admission.owner_token or type(admission.fenced) is not bool or not admission.fenced):
            raise Refused('exact fenced fixture custody required')
        process=admission.server_process
        if (type(process) is not subprocess.Popen or process.poll() is not None or type(process.args) is not list
                or '-dev' not in process.args or '-dev-no-store-token' not in process.args
                or '-dev-listen-address=127.0.0.1:'+str(admission.port) not in process.args):
            raise Refused('live owner-spawned disposable dev process required')
        self.admission=admission
        self._initialize_receipt(directory,operation_id,kid=kid,expected_version=expected_version,bindings=bindings)

    def _initialize_receipt(self,directory,operation_id,*,kid,expected_version,bindings):
        if (type(operation_id) is not str or not re.fullmatch(r'[a-zA-Z0-9_-]{1,80}',operation_id)
                or type(kid) is not str or not re.fullmatch(r'[a-zA-Z0-9_-]{1,80}',kid)
                or type(expected_version) is not int or not 0<=expected_version<2**31
                or type(bindings) is not dict or set(bindings)!=BINDINGS
                or any(type(v) is not str or not re.fullmatch(r'sha256:[0-9a-f]{64}',v) for v in bindings.values())):
            raise Refused('exact signer operation contract required')
        if directory.absolute()!=directory.resolve() or directory.is_symlink():raise Refused('indirect receipt custody refused')
        Journal._private(directory.stat(),directory=True)
        self.directory=directory;self.path=directory/(operation_id+'.json')
        self.expected_version=expected_version;self.kid=kid;self.attempted=False
        self.receipt={'schema':'railiance-platform.signer-cas-receipt.v1','operation_id':operation_id,
            'bindings':dict(bindings),'writer_sha256':'sha256:'+hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
            'actor':'synthetic-owner-fixture','created_at':datetime.now(timezone.utc).isoformat(),'path':PATH,'expected_version':expected_version,'kid':kid,
            'status':'prepared','production_admission':False,'rollback':'forward-recovery-only',
            'public_key':None,'public_fingerprint':None,'kv_version':None,'exact_readback':False}
        self._save(initial=True)

    def _save(self,*,initial=False):
        raw=encoded(self.receipt)
        if len(raw)>65536:raise Refused('receipt bound exceeded')
        if initial:
            target=self.path
        else:
            fd=os.open(self.path,os.O_RDONLY|os.O_NOFOLLOW|os.O_NONBLOCK)
            try:Journal._private(os.fstat(fd))
            finally:os.close(fd)
            target=self.directory/('.pending-'+secrets.token_hex(16))
        fd=os.open(target,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
        try:
            with os.fdopen(fd,'wb',closefd=False) as stream:stream.write(raw);stream.flush();os.fsync(fd)
        finally:os.close(fd)
        if not initial:os.replace(target,self.path)
        fd=os.open(self.directory,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW)
        try:os.fsync(fd)
        finally:os.close(fd)

    def _http(self,method,path,body=None):
        # Fixed loopback owner process and fixed data path; never arbitrary URL.
        if self.admission.server_process.poll() is not None:raise Refused('fixture custody backend unavailable')
        with _absolute_budget(2.0):
            conn=http.client.HTTPConnection('127.0.0.1',self.admission.port,timeout=2)
            try:
                conn.request(method,'/v1/'+path,body=None if body is None else encoded(body),headers={
                    'X-Vault-Token':self.admission.owner_token,'Content-Type':'application/json'})
                response=conn.getresponse();raw=response.read(1048577)
                if len(raw)>1048576:raise Refused('backend response bound exceeded')
                value=json.loads(raw,object_pairs_hook=_json_unique,parse_constant=_invalid_constant) if raw else {}
                return response.status,value
            finally:conn.close()

    def write(self):
        if self.attempted:raise Refused('signer operation already attempted')
        self.attempted=True
        # All source/receipt checks precede generation; no seed import, argv,
        # environment, temporary file or public output interface exists.
        private=Ed25519PrivateKey.generate()
        seed=b64(private.private_bytes(serialization.Encoding.Raw,serialization.PrivateFormat.Raw,serialization.NoEncryption()))
        public=b64(private.public_key().public_bytes(serialization.Encoding.Raw,serialization.PublicFormat.Raw))
        document={'ED25519_PRIVATE_KEY':seed,'ED25519_PUBLIC_KEY':public,'kid':self.kid}
        self.receipt.update(status='write-intent-unknown',public_key=public,
                            public_fingerprint='sha256:'+hashlib.sha256(public.encode()).hexdigest())
        self._save() # A durable intent must exist before any possible effect.
        try:
            status,result=self._http('POST',PATH,{'options':{'cas':self.expected_version},'data':document})
            if status!=200 or type(result) is not dict:raise Refused('signer write unconfirmed')
            version=result.get('data',{}).get('version')
            if type(version) is not int or version!=self.expected_version+1:raise Refused('signer write version unconfirmed')
            self.receipt['kv_version']=version
            status,result=self._http('GET',PATH+'?version='+str(version))
            if status!=200 or type(result) is not dict:raise Refused('signer readback unconfirmed')
            data=result.get('data',{});stored=data.get('data') if type(data) is dict else None
            metadata=data.get('metadata') if type(data) is dict else None
            if (type(stored) is not dict or set(stored)!=FIELDS or any(type(v) is not str for v in stored.values())
                    or type(metadata) is not dict or type(metadata.get('version')) is not int or metadata['version']!=version
                    or any(not hmac.compare_digest(stored[name],document[name]) for name in FIELDS)):
                raise Refused('signer readback mismatch')
            self.receipt.update(status='custody-verified-pending-independent-public-admission',exact_readback=True)
            self._save()
            return dict(self.receipt)
        except BaseException:
            # Persist only closed metadata; unknown never implies absence or retry.
            # If receipt storage fails, the durable earlier intent remains unknown.
            self.receipt.update(status='write-intent-unknown',exact_readback=False)
            self._save()
            raise Refused('signer custody outcome unconfirmed') from None


if __name__=='__main__':
    raise SystemExit('production signer custody unadmitted; no live entry point')