"""Offline outbound-webhook policy model. No DNS queries or sockets are opened.

This deliberately conservative fixture is not a complete HTTP transport.
It uses explicit network rules rather than version-dependent is_global behavior.
Run: python3 experiment.py
"""
import ipaddress
import json
import platform
from pathlib import Path
from urllib.parse import urlsplit, urljoin

DENY4 = [ipaddress.ip_network(n) for n in (
    '0.0.0.0/8', '10.0.0.0/8', '100.64.0.0/10', '127.0.0.0/8',
    '169.254.0.0/16', '172.16.0.0/12', '192.0.0.0/24',
    '192.0.2.0/24', '192.88.99.0/24', '192.168.0.0/16',
    '198.18.0.0/15', '198.51.100.0/24', '203.0.113.0/24',
    '224.0.0.0/4', '240.0.0.0/4')]
DENY6 = [ipaddress.ip_network(n) for n in (
    '2001::/23', '2001:db8::/32', '2002::/16', '3fff::/20',
    '2620:4f:8000::/48')]
GLOBAL6 = ipaddress.ip_network('2000::/3')
DNS = {
    'receiver.example': ['8.8.8.8'],
    'dual.example': ['8.8.8.8', '2606:4700:4700::1111'],
    'mixed.example': ['8.8.8.8', '10.0.0.8'],
    'private.example': ['10.0.0.8'],
    'empty.example': [],
    'redirect.example': ['9.9.9.9'],
    'rebind.example': ['8.8.8.8'],
}

class Denied(ValueError):
    pass

def permitted(address):
    try:
        ip = ipaddress.ip_address(address)
    except ValueError:
        return False
    if ip.version == 4:
        return not any(ip in network for network in DENY4)
    # Default deny outside this fixture's supported global-unicast envelope.
    return ip in GLOBAL6 and not any(ip in network for network in DENY6)

def plan(url, records=DNS):
    if any(ord(c) <= 32 or ord(c) >= 127 for c in url) or '\\' in url:
        raise Denied('unsupported URL characters')
    parsed = urlsplit(url)
    if parsed.scheme != 'https':
        raise Denied('HTTPS required')
    if parsed.username is not None or parsed.password is not None:
        raise Denied('URL credentials forbidden')
    if '#' in url:
        raise Denied('fragment forbidden')
    try:
        port = parsed.port
    except ValueError as error:
        raise Denied('invalid port') from error
    if port not in (None, 443):
        raise Denied('port forbidden')
    host = parsed.hostname
    if not host or host.endswith('.') or '%' in host:
        raise Denied('unsupported hostname')
    try:
        literal = ipaddress.ip_address(host)
        answers = [host]
    except ValueError:
        literal = None
        # Names are fixtures, not an implementation of DNS or IDNA handling.
        answers = records.get(host, [])
    if not answers:
        raise Denied('no supported address')
    if not all(permitted(address) for address in answers):
        raise Denied('address policy denied')
    authority = f'[{host}]' if literal and literal.version == 6 else host
    return dict(hostname=host, connectAddress=answers[0], port=443,
                tlsServerName=None if literal else host,
                certificateReference=host,
                certificateReferenceType='ip' if literal else 'dns',
                httpAuthority=authority, answers=answers)

CASES = [
    ('public IPv4', 'https://receiver.example/hook', True),
    ('public dual stack', 'https://dual.example/hook', True),
    ('loopback literal', 'https://127.0.0.1/hook', False),
    ('private hostname', 'https://private.example/hook', False),
    ('mixed DNS answer', 'https://mixed.example/hook', False),
    ('cloud metadata', 'https://169.254.169.254/hook', False),
    ('IPv6 loopback', 'https://[::1]/hook', False),
    ('mapped IPv4', 'https://[::ffff:8.8.8.8]/hook', False),
    ('documentation IPv6', 'https://[2001:db8::8]/hook', False),
    ('carrier NAT', 'https://100.64.0.1/hook', False),
    ('HTTP scheme', 'http://receiver.example/hook', False),
    ('URL credentials', 'https://user:secret@receiver.example/hook', False),
    ('fragment', 'https://receiver.example/hook#part', False),
    ('alternate port', 'https://receiver.example:8443/hook', False),
    ('empty DNS', 'https://empty.example/hook', False),
    ('ambiguous backslash', 'https://receiver.example\\@private.example/hook', False),
]

rows = []
for name, url, expected in CASES:
    try:
        connection = plan(url)
        actual = True
        reason = 'allowed connection plan'
    except (Denied, ValueError) as error:
        connection = None
        actual = False
        reason = str(error)
    assert actual == expected, name
    rows.append(dict(case=name, allowed=actual, reason=reason, plan=connection))

# A vulnerable transport ignores the validation answer and resolves again.
checked = plan('https://rebind.example/hook')
later_dns_answer = '10.0.0.8'
assert checked['connectAddress'] == '8.8.8.8'
assert not permitted(later_dns_answer)
rebind = dict(validationAddress=checked['connectAddress'],
              vulnerableConnectAddress=later_dns_answer,
              pinnedConnectAddress=checked['connectAddress'],
              pinnedTlsServerName=checked['tlsServerName'])

redirects = []
for location, expected in [('https://private.example/hook', False),
                           ('https://redirect.example/hook', True)]:
    target = urljoin('https://receiver.example/hook', location)
    try:
        checked_hop = plan(target)
        actual = True
    except Denied:
        checked_hop = None
        actual = False
    assert actual == expected
    redirects.append(dict(location=location, allowed=actual, plan=checked_hop))

# IP literals use IP certificate identity, no DNS SNI, and bracketed IPv6 authority.
literal_plans = [plan('https://8.8.8.8/hook'),
                 plan('https://[2606:4700:4700::1111]/hook')]
assert all(p['tlsServerName'] is None and p['certificateReferenceType'] == 'ip'
           for p in literal_plans)
assert literal_plans[1]['httpAuthority'] == '[2606:4700:4700::1111]'

result = dict(scope='Offline policy and connection-plan model; no live DNS, TLS or HTTP',
              pythonVersion=platform.python_version(), urlCaseCount=len(rows),
              allowedCases=sum(row['allowed'] for row in rows),
              deniedCases=sum(not row['allowed'] for row in rows),
              cases=rows, dnsRebinding=rebind, redirects=redirects,
              ipLiteralPlans=literal_plans,
              limitations=['No complete special-purpose registry implementation',
                           'No network transport, TLS handshake or proxy test',
                           'No redirect method, credential or retry semantics test'])
Path(__file__).with_name('results.json').write_text(json.dumps(result, indent=2) + '\n')
print(json.dumps({k: result[k] for k in ['urlCaseCount','allowedCases','deniedCases']}))
