import io
import os
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import unittest
import urllib.error
from scripts.verify_hosted_mcp import probe, origin, NoRedirect, credential_free_opener

BASE = 'https://crm.example'


class Reply(io.BytesIO):
    def __init__(self, path, body, status=200, headers=None):
        super().__init__(json.dumps(body).encode())
        self.status = status
        self.headers = headers or {}
        self.url = BASE + path

    def geturl(self):
        return self.url


class Opener:
    def __init__(self, bad_path=None, status=404):
        self.bad_path = bad_path
        self.status = status
        self.requests = []

    def open(self, request, timeout):
        self.requests.append(request)
        path = request.full_url.removeprefix(BASE)
        if path == self.bad_path:
            raise urllib.error.HTTPError(request.full_url, self.status, 'sensitive failure', {}, io.BytesIO(b'secret body'))
        if path == '/mcp':
            raise urllib.error.HTTPError(request.full_url, 401, 'Unauthorized', {
                'WWW-Authenticate': 'Bearer realm="mcp"'}, io.BytesIO())
        body = {'resource': BASE + '/mcp', 'bearer_methods_supported': ['header']}
        return Reply(path, body)


class ProbeTests(unittest.TestCase):
    def test_entry_pass_never_claims_authenticated_read(self):
        opener = Opener()
        result = probe(BASE, opener=opener)
        self.assertTrue(result['entryPassed'])
        self.assertEqual(result['authenticatedRead'], 'UNVERIFIED')
        self.assertEqual(result['authenticatedWrite'], 'UNVERIFIED')
        self.assertFalse(any('authorization-server' in r.full_url for r in opener.requests))
        self.assertEqual(len(opener.requests), 2)
        self.assertTrue(all(not r.has_header('Authorization') for r in opener.requests))

    def test_each_missing_or_redirected_endpoint_fails_without_body_leak(self):
        for path in ('/mcp', '/.well-known/oauth-protected-resource'):
            for status in (200, 302, 404, 500):
                with self.subTest(path=path, status=status):
                    result = probe(BASE, opener=Opener(path, status))
                    self.assertFalse(result['entryPassed'])
                    self.assertEqual(len(result['checks']), 2)
                    self.assertNotIn('secret body', json.dumps(result))
                    self.assertNotIn('sensitive failure', json.dumps(result))

    def test_discovery_wrong_audience_and_invalid_json_fail(self):
        class Invalid(Opener):
            def open(self, request, timeout):
                if request.full_url.endswith('protected-resource'):
                    return Reply('/.well-known/oauth-protected-resource', {'resource': BASE + '/other'})
                return super().open(request, timeout)
        self.assertFalse(probe(BASE, opener=Invalid())['entryPassed'])

    def test_redirect_handler_never_creates_a_followup_request(self):
        self.assertIsNone(NoRedirect().redirect_request(None, None, 302, '', {}, BASE + '/login'))

    def test_oversized_discovery_and_transport_failure_are_sanitized(self):
        class Invalid(Opener):
            def open(self, request, timeout):
                if request.full_url.endswith('protected-resource'):
                    return Reply('/.well-known/oauth-protected-resource', 'x' * 65537)
                raise urllib.error.URLError('private transport detail')
        result = probe(BASE, opener=Invalid())
        self.assertFalse(result['entryPassed'])
        self.assertEqual(result['checks'][0]['reason'], 'invalid discovery JSON')
        self.assertNotIn('private transport detail', json.dumps(result))

    def test_ambient_proxy_credentials_never_reach_proxy_or_target(self):
        class Handler(BaseHTTPRequestHandler):
            def do_GET(self):
                self.server.requests.append(self.headers.get('Proxy-Authorization'))
                self.send_response(200)
                self.end_headers()
            def log_message(self, *args):
                pass
        target = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
        proxy = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
        servers = (target, proxy)
        for server in servers:
            server.requests = []
            threading.Thread(target=server.serve_forever, daemon=True).start()
        proxy_url = f'http://fixture-user:fixture-password@127.0.0.1:{proxy.server_port}'
        changes = {name: proxy_url for name in ('HTTP_PROXY', 'http_proxy', 'HTTPS_PROXY', 'https_proxy', 'ALL_PROXY', 'all_proxy')}
        changes.update({'NO_PROXY': '', 'no_proxy': ''})
        before = {name: os.environ.get(name) for name in changes}
        try:
            os.environ.update(changes)
            opener = credential_free_opener()
            with opener.open(f'http://127.0.0.1:{target.server_port}/probe', timeout=2) as response:
                self.assertEqual(response.status, 200)
            self.assertEqual(target.requests, [None])
            self.assertEqual(proxy.requests, [])
            self.assertFalse(any(isinstance(handler, urllib.request.ProxyHandler) and handler.proxies for handler in opener.handlers))
        finally:
            for name, value in before.items():
                if value is None:
                    os.environ.pop(name, None)
                else:
                    os.environ[name] = value
            for server in servers:
                server.shutdown()
                server.server_close()

    def test_rejects_noncanonical_or_credentialed_targets(self):
        for target in ('http://crm.example', BASE + '/mcp', 'https://user:password@crm.example', BASE + '?token=secret', BASE + '#x', 'https://crm.example:bad'):
            with self.subTest(target=target):
                with self.assertRaises(ValueError):
                    origin(target)


if __name__ == '__main__':
    unittest.main()