railiance-platform/scripts/openbao_policy_sync.py

#!/usr/bin/env python3
"""Silent contained write of one declared ACL policy, guarded against drift.

Writes openbao/policies/<name>.hcl to sys/policy/<name> only when the live
rules still equal the previously declared version (--expect-live-sha256). If
the live rules already equal the file, nothing is written. Any other live
state is refused, so an undeclared live edit is never overwritten. Readback
must match. One non-secret receipt; no output.

  python3 scripts/openbao-attended-exec.py -- scripts/openbao-policy-sync.sh \\
    --policy <name> --expect-live-sha256 <sha> --receipt <absolute file>
"""
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import re
import subprocess
import sys

POLICY_DIR = Path(__file__).resolve().parents[1] / 'openbao/policies'
NAME = re.compile(r'^[a-z0-9][a-z0-9-]{1,80}$')


class Refused(Exception):
    pass


def digest(text):
    return hashlib.sha256(text.strip().encode('utf-8')).hexdigest()


def require_attended():
    if Path.home().parent.name != '.warden-attended-login' or os.getenv('BAO_TOKEN') or os.getenv('VAULT_TOKEN'):
        raise Refused('attended_envelope_required')


def read_live(name):
    result = subprocess.run(['bao', 'read', '-format=json', 'sys/policy/' + name],
                            capture_output=True, check=True, timeout=30)
    return json.loads(result.stdout)['data']['rules']


def write_live(name, path):
    subprocess.run(['bao', 'policy', 'write', name, str(path)],
                   capture_output=True, check=True, timeout=30)


def sync(name, expect, read=read_live, write=write_live):
    path = POLICY_DIR / (name + '.hcl')
    declared = digest(path.read_text(encoding='utf-8'))
    live = digest(read(name))
    if live == declared:
        return {'status': 'already_current', 'changed': False, 'live_sha256': live}
    if live != expect:
        raise Refused('live_policy_drifted')
    write(name, path)
    after = digest(read(name))
    if after != declared:
        raise Refused('readback_mismatch')
    return {'status': 'applied', 'changed': True, 'previous_sha256': live, 'live_sha256': after}


def write_receipt(path, name, **body):
    fd = os.open(Path(path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    body = {
        'schema': 'railiance-platform.openbao-policy-sync.v1',
        'observed_at': datetime.now(timezone.utc).isoformat(),
        'policy_name': name, 'credential_values_emitted': False, **body,
    }
    with os.fdopen(fd, 'w', encoding='utf-8') as handle:
        json.dump(body, handle, indent=2, sort_keys=True)
        handle.write('\n')


def classify(error):
    if isinstance(error, Refused):
        return str(error)
    if isinstance(error, subprocess.CalledProcessError):
        return 'bao_write_failed' if 'policy' in error.cmd else 'bao_read_failed'
    return 'contained_operation_failed'


def parse(argv):
    opts, args = {}, list(argv)
    while args:
        if args[0] in ('--policy', '--expect-live-sha256', '--receipt') and len(args) > 1:
            opts[args[0]] = args[1]
            args = args[2:]
        else:
            raise SystemExit(2)
    if set(opts) != {'--policy', '--expect-live-sha256', '--receipt'} or not NAME.match(opts['--policy']):
        raise SystemExit(2)
    if not (POLICY_DIR / (opts['--policy'] + '.hcl')).is_file():
        raise SystemExit(2)
    return opts['--policy'], opts['--expect-live-sha256'], opts['--receipt']


def main(argv):
    name, expect, receipt = parse(argv)
    try:
        require_attended()
        result = sync(name, expect)
        write_receipt(receipt, name, declared_sha256=digest((POLICY_DIR / (name + '.hcl')).read_text()), **result)
        return 0
    except Exception as error:
        try:
            write_receipt(receipt, name, status=classify(error), changed=False)
        except Exception:
            pass
        return 1


if __name__ == '__main__':
    raise SystemExit(main(sys.argv[1:]))