tjh.dev
nixpkgs mirror (for testing)
github.com/NixOS/nixpkgs
nix
1{ lib, pkgs, ... }:
2let
3 # https://docs.mitmproxy.org/stable/concepts/certificates/#using-a-custom-certificate-authority
4 caCert = pkgs.runCommand "ca-cert" { } ''
5 touch $out
6 cat ${./common/acme/server/ca.key.pem} >> $out
7 cat ${./common/acme/server/ca.cert.pem} >> $out
8 '';
9in
10{
11 name = "mitmproxy";
12 meta.maintainers = [ lib.teams.ngi.members ];
13
14 nodes.machine =
15 { pkgs, ... }:
16 {
17 security.pki.certificateFiles = [ caCert ];
18
19 services.getty.autologinUser = "root";
20
21 environment.systemPackages =
22 let
23 # A counter. It has 2 functions:
24 # 1. GET /old and GET /new, to demostrate rewriting requests.
25 # 2. GET /counter and POST /counter, to demonstrate replaying requests.
26 counter = pkgs.writers.writePython3Bin "counter" { } ''
27 from http.server import BaseHTTPRequestHandler, HTTPServer
28
29 counter = 0
30
31
32 class HTTPRequestHandler(BaseHTTPRequestHandler):
33 def do_POST(self):
34 match self.path:
35 case "/counter":
36 global counter
37 counter += 1
38 self.send_response(204)
39 self.end_headers()
40 case _:
41 self.send_response(404)
42 self.end_headers()
43
44 def do_GET(self):
45 match self.path:
46 case "/counter":
47 self.send_response(200)
48 self.send_header("Content-type", "text/plain")
49 self.end_headers()
50 _ = self.wfile.write(str(counter).encode())
51 case "/old":
52 self.send_response(200)
53 self.send_header("Content-type", "text/plain")
54 self.end_headers()
55 _ = self.wfile.write("fail".encode())
56 case "/new":
57 self.send_response(200)
58 self.send_header("Content-type", "text/plain")
59 self.end_headers()
60 _ = self.wfile.write("success".encode())
61 case _:
62 self.send_response(404)
63 self.end_headers()
64
65
66 server_address = ("", 8000)
67 server = HTTPServer(server_address, HTTPRequestHandler)
68 server.serve_forever()
69 '';
70 in
71 [
72 counter
73 pkgs.mitmproxy
74 ];
75 };
76
77 testScript =
78 let
79 addonScript = pkgs.writeText "addon-script" ''
80 def request(flow):
81 # https://docs.mitmproxy.org/stable/api/mitmproxy/http.html#Request
82 flow.request.path = "/new"
83 '';
84 in
85 ''
86 def curl(command: str, proxy: bool = False):
87 if proxy:
88 command = "curl --proxy 127.0.0.1:8080 --cacert ~/.mitmproxy/mitmproxy-ca-cert.pem " + command
89 else:
90 command = "curl " + command
91 return machine.succeed(command)
92
93 start_all()
94 machine.wait_for_unit("default.target")
95
96 # https://docs.mitmproxy.org/stable/concepts/certificates/#using-a-custom-certificate-authority
97 machine.succeed("mkdir -p ~/.mitmproxy")
98 machine.succeed("ln -s ${caCert} ~/.mitmproxy/mitmproxy-ca.pem")
99
100 machine.succeed("counter >/dev/null &")
101 machine.wait_for_open_port(8000)
102
103 # rewrite
104 # https://docs.mitmproxy.org/stable/mitmproxytutorial-modifyrequests/
105
106 t.assertEqual("fail", curl("http://localhost:8000/old"))
107
108 machine.send_chars("mitmdump -s ${addonScript}\n")
109 machine.wait_for_open_port(8080)
110
111 t.assertEqual("success", curl("http://localhost:8000/old", proxy=True))
112
113 machine.send_key("ctrl-c")
114
115 # replay
116 # https://docs.mitmproxy.org/stable/mitmproxytutorial-replayrequests/
117 # https://docs.mitmproxy.org/stable/tutorials/client-replay/
118
119 t.assertEqual("0", curl("http://localhost:8000/counter"))
120
121 machine.send_chars("mitmdump -w replay\n")
122 machine.wait_for_open_port(8080)
123
124 curl("-X POST http://localhost:8000/counter", proxy=True)
125
126 machine.send_key("ctrl-c")
127
128 t.assertEqual("1", curl("http://localhost:8000/counter"))
129
130 machine.succeed("mitmdump -C /root/replay")
131
132 t.assertEqual("2", curl("http://localhost:8000/counter"))
133 '';
134}