coulomb-social/scripts/rehearse-mail-return-boundary.py

#!/usr/bin/env python3
"""Exercise owner mail sources using temporary ledgers and synthetic mail only.

This probe documents present capabilities and missing integration. It is not an
inbound worker, sender deployment, credential admission or suppression operator.
"""
from __future__ import annotations

import argparse
import hashlib
import io
import json
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import patch


def main():
    if not __debug__:
        raise SystemExit('Assertions must be enabled')
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--owner-repo', type=Path, required=True)
    parser.add_argument('--native-evidence', type=Path, required=True)
    args = parser.parse_args()
    source = args.owner_repo.resolve() / 'src'
    native = json.loads(args.native_evidence.read_text())
    module = source / 'email_connect/transactional.py'
    assert hashlib.sha256(module.read_bytes()).hexdigest() == native[
        'running_process_configuration']['transactional_source_sha256']
    # Fail any attempted Python network connection/bind, even if a future source
    # change accidentally substitutes a real transport for these fixtures.
    def no_network(event, _args):
        if event in {'socket.connect', 'socket.bind', 'socket.getaddrinfo', 'socket.sendto'}:
            raise RuntimeError('Offline rehearsal forbids network access')
    sys.addaudithook(no_network)
    sys.path.insert(0, str(source))
    from email_connect.callers import CallerPolicy, ScopedTransactionalApplication
    from email_connect.config import (AppConfig, MailboxConfig, ReportsConfig,
                                      ScanConfig, SourceConfig, StorageConfig)
    from email_connect.scanner import scan_mailbox
    from email_connect.storage import StateStore
    from email_connect.transactional import SMTPProvider, SQLiteDeliveryStore

    captured = []
    class SyntheticSMTP:
        def __init__(self, *args, **kwargs):
            self.tls = False
        def __enter__(self):
            return self
        def __exit__(self, *args):
            pass
        def starttls(self, context):
            assert context.check_hostname
            self.tls = True
        def login(self, username, password):
            assert self.tls and username == 'fixture' and password == 'fixture-only'
        def send_message(self, message):
            assert message['To'] == 'recipient@example.test'
            captured.append(message)

    def policy(name, operations, tenant='tenant:fixture'):
        return CallerPolicy(caller_id=name, token='fixture-credential-' + name,
                            source=name, bindings=frozenset({(tenant, 'client:fixture')}),
                            operations=frozenset(operations),
                            templates=frozenset({'family_member.invited'}),
                            rate_per_minute=60, burst=30, max_payload_bytes=8192)
    sender = policy('fixture-sender', {'send', 'delivery-status'})
    operator = policy('fixture-operator', {'suppress', 'unsuppress'})
    other = policy('fixture-other', {'send'}, 'tenant:other')
    def event(name, caller=sender, tenant='tenant:fixture'):
        return {'id': name, 'source': caller.source, 'type': 'family_member.invited',
                'tenant': tenant, 'client_id': 'client:fixture',
                'data': {'primary_email': 'recipient@example.test', 'invitation_id': name}}
    def invoke(app, payload, path='/v1/send', caller=sender):
        raw = json.dumps(payload).encode(); result = {}
        env = {'PATH_INFO': path, 'REQUEST_METHOD': 'POST', 'CONTENT_LENGTH': str(len(raw)),
               'wsgi.input': io.BytesIO(raw), 'HTTP_AUTHORIZATION': 'Bearer ' + caller.token,
               'HTTP_IDEMPOTENCY_KEY': payload.get('id', '')}
        body = b''.join(app(env, lambda status, headers: result.update(status=int(status[:3]))))
        return result['status'], json.loads(body)

    with tempfile.TemporaryDirectory(prefix='csoc-mail-return-') as tmp, patch(
            'email_connect.transactional.smtplib.SMTP', SyntheticSMTP):
        root = Path(tmp)
        ledger = SQLiteDeliveryStore(str(root / 'transactional.sqlite'))
        provider = SMTPProvider('smtp.example.test', 587, 'fixture', 'fixture-only',
                                'platform@example.test')
        app = ScopedTransactionalApplication(ledger, provider, 'https://issuer.example.test',
                                             [sender, operator, other])
        assert invoke(app, event('invitation-1'))[0] == 202
        assert len(captured) == 1
        message = captured[0]
        reference = str(message['Message-ID'])
        assert message['From'] == 'platform@example.test'
        assert message['Reply-To'] is None and message['Return-Path'] is None
        assert ledger.db.execute('SELECT count(*) FROM deliveries WHERE provider_ref=?',
                                 (reference,)).fetchone()[0] == 1
        fixture = root / 'mailbox'; fixture.mkdir()
        (fixture / 'bounce.eml').write_text(
            'From: mailer-daemon@example.test\nTo: platform@example.test\n'
            'Subject: Delivery Status Notification (Failure)\n'
            'Date: Sat, 31 Oct 2026 10:00:00 +0000\n'
            'Message-ID: <fixture-bounce@example.test>\n'
            'Content-Type: text/plain; charset=utf-8\n\n'
            'Delivery failure.\nFinal-Recipient: rfc822; recipient@example.test\n'
            'Action: failed\nStatus: 5.1.1\nDiagnostic-Code: smtp; 550 User unknown\n'
            'Original-Message-ID: ' + reference + '\n')
        config = AppConfig(mailbox=MailboxConfig(id='fixture-mailbox', protocol='fixture'),
                           scan=ScanConfig(), storage=StorageConfig(str(root / 'scanner.sqlite')),
                           reports=ReportsConfig(str(root / 'reports')),
                           source=SourceConfig(fixture_dir=str(fixture)))
        scan = scan_mailbox(config)
        evidence = StateStore(config.storage.path)
        rows = evidence.evidence_rows()
        assert scan.scan.messages_seen == 1 and len(rows) == 1
        assert rows[0]['original_message_id'] == reference
        assert rows[0]['affected_email_address'] == 'recipient@example.test'
        assert rows[0]['assessment_subclass'] == 'fail.hard_bounce'
        assert not ledger.is_suppressed('recipient@example.test')
        repeat = scan_mailbox(config)
        assert repeat.scan.messages_new == 0 and repeat.scan.evidence_events_created == 0
        payload = {'recipient': 'recipient@example.test', 'reason': 'hard_bounce'}
        assert invoke(app, payload, '/v1/suppressions')[0] == 403
        assert not ledger.is_suppressed('recipient@example.test')
        assert invoke(app, payload, '/v1/suppressions', operator)[0] == 200
        assert invoke(app, event('invitation-2')) == (400, {'error': 'recipient_suppressed'})
        assert invoke(app, event('invitation-3', other, 'tenant:other'), caller=other) == (
            400, {'error': 'recipient_suppressed'})
        assert len(captured) == 1
        ledger.db.close()
        ledger = SQLiteDeliveryStore(str(root / 'transactional.sqlite'))
        assert ledger.is_suppressed('recipient@example.test')
        app = ScopedTransactionalApplication(ledger, provider, 'https://issuer.example.test',
                                             [sender, operator, other])
        assert invoke(app, payload, '/v1/suppressions/remove')[0] == 403
        assert invoke(app, payload, '/v1/suppressions/remove', operator)[0] == 200
        assert not ledger.is_suppressed('recipient@example.test')
        assert len(captured) == 1
        ledger.db.close(); evidence.close()
    assert not root.exists()
    print(json.dumps({
        'schema': 'coulomb.mail-return-boundary-rehearsal.v1',
        'completed_at': datetime.now(timezone.utc).isoformat(),
        'source_sha256': {str(p.relative_to(args.owner_repo.resolve())):
                          hashlib.sha256(p.read_bytes()).hexdigest()
                          for p in sorted((source / 'email_connect').glob('*.py'))},
        'probe_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        'transactional_source_matches_native': True,
        'smtp_transport': 'in-process synthetic replacement; no SMTP connection',
        'message_id_correlates_with_ledger_and_synthetic_bounce': True,
        'explicit_reply_to_or_return_path_headers': False,
        'scanner_deduplicates_evidence': True,
        'scanner_updates_transactional_suppression': False,
        'ordinary_sender_suppress_and_unsuppress_denied': True,
        'synthetic_operator_suppression_blocks_send_and_survives_restart': True,
        'suppression_scope': 'recipient-global across callers and tenants',
        'explicit_synthetic_operator_unsuppression_verified': True,
        'native_operator_admission_proved': False,
        'received_mail_authenticity_or_correlation_authority_proved': False,
        'python_network_operations_blocked': True,
        'temporary_ledgers_removed': True,
        'mail_sent': False, 'production_mailbox_accessed': False,
    }, indent=2))


if __name__ == '__main__':
    main()