ivoryvalley 0.1.0

A transparent deduplication proxy for Mastodon and the Fediverse
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

#!/usr/bin/env python3
"""
Simple HTTPS reverse proxy for local development.
Forwards https://localhost:8443 -> http://localhost:8080
"""

import http.server
import http.client
import ssl
import sys
import os

LISTEN_PORT = 443
UPSTREAM_HOST = "localhost"
UPSTREAM_PORT = 8080


class ProxyHandler(http.server.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def do_proxy(self):
        # Read request body if present
        content_length = self.headers.get('Content-Length')
        body = None
        if content_length:
            body = self.rfile.read(int(content_length))

        # Connect to upstream
        conn = http.client.HTTPConnection(UPSTREAM_HOST, UPSTREAM_PORT)

        # Build headers (pass through all except hop-by-hop)
        headers = {}
        hop_by_hop = ('connection', 'keep-alive', 'transfer-encoding',
                      'te', 'trailer', 'upgrade', 'proxy-authorization',
                      'proxy-authenticate')
        for key, value in self.headers.items():
            if key.lower() not in hop_by_hop:
                headers[key] = value

        try:
            conn.request(self.command, self.path, body=body, headers=headers)
            resp = conn.getresponse()

            # Send response status
            self.send_response_only(resp.status, resp.reason)

            # Forward all response headers
            # Note: getheaders() combines Set-Cookie headers, we need them separate
            for key, value in resp.getheaders():
                if key.lower() == 'set-cookie':
                    # Handle multiple cookies (they come combined with ', ' but that breaks parsing)
                    # Actually resp.msg.get_all() gives us the raw headers
                    continue  # Handle below
                elif key.lower() not in ('transfer-encoding', 'connection'):
                    self.send_header(key, value)

            # Handle Set-Cookie headers separately (each must be its own header)
            if hasattr(resp, 'msg') and resp.msg:
                cookies = resp.msg.get_all('Set-Cookie') or []
                for cookie in cookies:
                    self.send_header('Set-Cookie', cookie)

            self.send_header('Connection', 'close')
            self.end_headers()

            # Forward response body
            self.wfile.write(resp.read())

        except Exception as e:
            self.send_error(502, f"Upstream error: {e}")
        finally:
            conn.close()

    def do_GET(self): self.do_proxy()
    def do_POST(self): self.do_proxy()
    def do_PUT(self): self.do_proxy()
    def do_DELETE(self): self.do_proxy()
    def do_PATCH(self): self.do_proxy()
    def do_OPTIONS(self): self.do_proxy()
    def do_HEAD(self): self.do_proxy()

    def log_message(self, format, *args):
        print(f"[HTTPS] {self.address_string()} - {format % args}")


def main():
    script_dir = os.path.dirname(os.path.abspath(__file__))
    cert_dir = os.path.join(script_dir, ".certs")
    cert_file = os.path.join(cert_dir, "localhost.crt")
    key_file = os.path.join(cert_dir, "localhost.key")

    if not os.path.exists(cert_file):
        print(f"Error: Certificate not found at {cert_file}")
        print("Run the dev-https.sh script first to generate certificates.")
        sys.exit(1)

    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.load_cert_chain(cert_file, key_file)

    server = http.server.HTTPServer(("0.0.0.0", LISTEN_PORT), ProxyHandler)
    server.socket = context.wrap_socket(server.socket, server_side=True)

    print(f"HTTPS proxy running on https://localhost")
    print(f"Forwarding to http://{UPSTREAM_HOST}:{UPSTREAM_PORT}")
    print("Press Ctrl+C to stop")

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down...")


if __name__ == "__main__":
    main()