railiance-platform/scripts/email_connect_smtp_delivery.py

#!/usr/bin/env python3
"""Fixed, silent email-connect platform-mail reader admission.

Default is offline proposal output. Execution requires the approved CCR, native
binding confirmation and reviewed source digest inside the attended owner lane.
It neither creates Kubernetes resources nor deploys the application sender.
"""
from __future__ import annotations

import argparse
import copy
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import re
import sys
from datetime import datetime, timezone

ROOT = Path(__file__).resolve().parents[1]
OPENBAO_CLUSTER_ID = 'fd28df5d-98ec-57dd-42ec-9b3e4f4e53bf'
CCR = ROOT / 'credential-change-requests/CCR-2026-0056-email-connect-platform-smtp.yaml'
POLICY_FILE = ROOT / 'openbao/policies/email-connect-platform-smtp-eso.hcl'
LANE = dict(service_account='email-connect-platform-smtp-eso', namespace='email-connect',
            role='email-connect-platform-smtp-eso', policy='email-connect-platform-smtp-eso',
            kv_path='platform/data/workloads/railiance-telemetry/smtp')
SOURCES = ('scripts/email_connect_smtp_delivery.py', 'scripts/telemetry_smtp_delivery.py',
           'scripts/state_hub_preflight_lane.py', 'scripts/repair_eso_kubernetes_auth.py',
           'scripts/telemetry_smtp_entry.py', 'scripts/credential-change.py',
           'openbao/policies/email-connect-platform-smtp-eso.hcl')
PROTECTED = ('sys/policies/acl/agent-high-risk-boundary',
             'sys/policies/acl/telemetry-smtp-eso', 'auth/kubernetes/role/telemetry-smtp-eso',
             'sys/policies/acl/authelia-smtp-eso', 'auth/kubernetes/role/authelia-smtp-eso')


class Refused(Exception):
    pass


def require(ok, reason):
    if not ok:
        raise Refused(reason)


def module(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    result = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(result)
    return result


def lane_module():
    sys.path.insert(0, str(ROOT / 'scripts'))
    lane = module('email_connect_smtp_fixed_lane', ROOT / 'scripts/telemetry_smtp_delivery.py')
    lane.LANE = dict(LANE)
    lane.POLICY = POLICY_FILE.read_text()
    return lane


def prepare():
    if not __debug__:
        raise Refused('assertions_must_be_enabled')
    validator = module('email_connect_smtp_ccr', ROOT / 'scripts/credential-change.py')
    ccr, errors, _ = validator.validate_ccr(CCR)
    require(not errors, 'invalid_ccr')
    lane = lane_module()
    expected_auth = dict(method='kubernetes', mount='kubernetes', role=LANE['role'],
                         bound_claims={'service_account_names': [LANE['service_account']],
                                       'service_account_namespaces': [LANE['namespace']]},
                         policies=[LANE['policy']], ttl='15m', audience='openbao',
                         token_max_ttl='15m', token_explicit_max_ttl='15m',
                         token_no_default_policy=True)
    actual = dict(ccr['openbao']['auth']); actual.pop('bound_claims_confirmed', None)
    require(actual == expected_auth, 'auth_contract_drift')
    expected_policy = ('path "'+LANE['kv_path']+'" { capabilities = ["read"] }\n'
                       'path "auth/token/lookup-self" { capabilities = ["read"] }\n'
                       'path "auth/token/revoke-self" { capabilities = ["update"] }\n')
    require(lane.POLICY == expected_policy, 'reader_policy_drift')
    o = ccr['openbao']
    require(o['kv_path'] == 'platform/workloads/railiance-telemetry/smtp'
            and o['mount'] == 'platform' and o['fields'] == ['SMTP_PASSWORD']
            and o['metadata_read'] is False and o['token_self_lifecycle'] is True
            and o['policy_name'] == LANE['policy']
            and o['policy_file'] == str(POLICY_FILE.relative_to(ROOT)), 'path_contract_drift')
    delivery = ccr['delivery']
    require(all(delivery.get(k) == v for k, v in {
        'namespace': 'email-connect', 'service_account': LANE['service_account'],
        'secret_store': 'openbao-email-connect-platform-smtp',
        'external_secret': 'email-connect-platform-smtp',
        'target_secret': 'email-connect-platform-smtp', 'target_key': 'password',
    }.items()), 'projection_contract_drift')
    scope = copy.deepcopy(ccr)
    for key in ('status', 'review', 'updated', 'state_hub'):
        scope.pop(key, None)
    scope['openbao']['auth'].pop('bound_claims_confirmed', None)
    scope['access_frontdoor'].pop('readiness', None)
    proposal = {'schema': 'email-connect.smtp-admission-proposal.v1', 'scope': scope,
                'openbao_cluster_id': OPENBAO_CLUSTER_ID,
                'role': lane.role_payload(LANE), 'policy': lane.POLICY,
                'source_sha256': {p: hashlib.sha256((ROOT / p).read_bytes()).hexdigest()
                                  for p in SOURCES}}
    raw = json.dumps(proposal, sort_keys=True, separators=(',', ':')).encode()
    return ccr, proposal, 'sha256:' + hashlib.sha256(raw).hexdigest()


def execution_guard(ccr, expected, actual):
    require(expected == actual, 'reviewed_digest_required')
    require(ccr['status'] == 'approved', 'approved_ccr_required')
    require(ccr['openbao']['auth'].get('bound_claims_confirmed') is True,
            'native_binding_confirmation_required')


def protected_state(lane):
    return {p: lane.data(lane.bao(['read', '-format=json', p]))['data'] for p in PROTECTED}


def admit(receipt, checkpoint):
    lane = lane_module()
    status = lane.data(lane.bao(['status', '-format=json']))
    require(status.get('cluster_id') == OPENBAO_CLUSTER_ID
            and status.get('initialized') is True and status.get('sealed') is False,
            'openbao_instance_mismatch')
    ident = lane.data(lane.bao(['token', 'lookup', '-format=json']))['data']['policies']
    require('platform-admin' in ident and 'root' not in ident, 'attended_platform_admin_required')
    before = protected_state(lane)
    boundary = before[PROTECTED[0]]['policy']
    for path in (LANE['kv_path'], LANE['kv_path'].replace('/data/', '/metadata/', 1)):
        require(re.search(r'path\s+"'+re.escape(path)+r'"\s*\{\s*capabilities\s*=\s*\["deny"\]\s*\}', boundary),
                'existing_agent_boundary_required')
    # Check both existing targets before the shared lane can create either one.
    for path, verify in (
        ('sys/policies/acl/'+LANE['policy'], lambda v: require(v['policy'] == lane.POLICY, 'reader_policy_drift')),
        ('auth/kubernetes/role/'+LANE['role'], lambda v: lane.check_role(v, LANE)),
    ):
        old = lane.bao(['read', '-format=json', path], allow_failure=True)
        if old.returncode == 0:
            verify(lane.data(old)['data'])
        else:
            require(lane.missing(old), 'metadata_absence_unproven')
    mailbox_before = lane.data(lane.bao(['read', '-format=json', LANE['kv_path']]))['data']
    lane.validate_entry(mailbox_before)
    receipt.update(phase='admitting', mutation_outcome='unconfirmed')
    checkpoint(receipt)
    lane.run(receipt)
    require(protected_state(lane) == before, 'existing_consumer_metadata_changed')
    mailbox_after = lane.data(lane.bao(['read', '-format=json', LANE['kv_path']]))['data']
    require(mailbox_after == mailbox_before, 'mailbox_changed_during_admission')
    receipt.update(status='reader_admitted_verified', phase='completed', mutation_outcome='verified',
                   existing_consumers_preserved=True, agent_boundary_preserved=True,
                   mailbox_unchanged=True, kubernetes_resources_installed=False,
                   native_consumer_auth_verified=False, projection_ready=False,
                   application_sender_changed=False)
    checkpoint(receipt)


def execute(ccr, proposal, digest, expected, path, runner=admit):
    execution_guard(ccr, expected, digest)
    parent = path.parent
    info = parent.lstat()
    require(parent.is_dir() and not parent.is_symlink() and info.st_uid == os.getuid()
            and not info.st_mode & 0o077, 'private_receipt_parent_required')
    fd = os.open(path, os.O_WRONLY | os.O_APPEND | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
    # This unique receipt is durable before any owner operation. An incomplete
    # receipt requires inspection; never overwrite it or automatically retry.
    receipt = {'schema': 'email-connect.smtp-admission.v1', 'status': 'started',
               'phase': 'preflight', 'proposal_digest': digest,
               'source_sha256': proposal['source_sha256'],
               'started_at': datetime.now(timezone.utc).isoformat(), 'email_sent': False,
               'password_modified': False, 'mutation_outcome': 'not_started'}
    def checkpoint(value):
        raw = (json.dumps(value, sort_keys=True, separators=(',', ':'))+'\n').encode()
        while raw:
            written = os.write(fd, raw)
            if written <= 0:
                raise OSError('receipt_write_failed')
            raw = raw[written:]
        os.fsync(fd)
    try:
        checkpoint(receipt)
        directory = os.open(parent, os.O_RDONLY | os.O_DIRECTORY)
        try:
            os.fsync(directory)
        finally:
            os.close(directory)
        runner(receipt, checkpoint)
        require(receipt.get('status') == 'reader_admitted_verified', 'success_receipt_required')
        result = 0
    except Exception as error:
        # Exceptions may carry provider text. Only this wrapper's fixed guards
        # are permitted diagnostics; dependencies and arbitrary errors stay opaque.
        safe = {'openbao_instance_mismatch', 'attended_platform_admin_required', 'existing_agent_boundary_required',
                'reader_policy_drift', 'metadata_absence_unproven',
                'existing_consumer_metadata_changed', 'mailbox_changed_during_admission',
                'success_receipt_required'}
        receipt.update(status='failed_requires_inspection', reason=(str(error)
                       if isinstance(error, Refused) and str(error) in safe
                       else 'contained_admission_failure'))
        result = 1
    finally:
        time_key = ('completed_at' if receipt['status'] in
                    {'reader_admitted_verified', 'failed_requires_inspection'} else 'interrupted_at')
        receipt[time_key] = datetime.now(timezone.utc).isoformat()
        checkpoint(receipt)
        os.close(fd)
    return result


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--execute', action='store_true')
    parser.add_argument('--expect-digest')
    parser.add_argument('--receipt', type=Path)
    args = parser.parse_args(argv)
    ccr, proposal, digest = prepare()
    if not args.execute:
        print(json.dumps({'status': 'prepared', 'proposal_digest': digest,
                          'ccr_status': ccr['status'],
                          'binding_confirmed': ccr['openbao']['auth']['bound_claims_confirmed'],
                          'proposal': proposal}, indent=2))
        return 0
    require(args.receipt is not None, 'private_receipt_required')
    return execute(ccr, proposal, digest, args.expect_digest, args.receipt)


if __name__ == '__main__':
    if '--execute' in sys.argv:
        with open(os.devnull, 'w') as sink:
            os.dup2(sink.fileno(), 1); os.dup2(sink.fileno(), 2)
            try:
                code = main()
            except Exception:
                code = 1
        sys.exit(code)
    sys.exit(main())