pyrox.dev
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 pkgs.mitmproxy2swagger
75 ];
76 };
77
78 testScript =
79 let
80 addonScript = pkgs.writeText "addon-script" ''
81 def request(flow):
82 # https://docs.mitmproxy.org/stable/api/mitmproxy/http.html#Request
83 flow.request.path = "/new"
84 '';
85 in
86 ''
87 def curl(command: str, proxy: bool = False):
88 if proxy:
89 command = "curl --proxy 127.0.0.1:8080 --cacert ~/.mitmproxy/mitmproxy-ca-cert.pem " + command
90 else:
91 command = "curl " + command
92 return machine.succeed(command)
93
94 start_all()
95 machine.wait_for_unit("default.target")
96
97 # https://docs.mitmproxy.org/stable/concepts/certificates/#using-a-custom-certificate-authority
98 machine.succeed("mkdir -p ~/.mitmproxy")
99 machine.succeed("ln -s ${caCert} ~/.mitmproxy/mitmproxy-ca.pem")
100
101 machine.succeed("counter >/dev/null &")
102 machine.wait_for_open_port(8000)
103
104 # rewrite
105 # https://docs.mitmproxy.org/stable/mitmproxytutorial-modifyrequests/
106
107 t.assertEqual("fail", curl("http://localhost:8000/old"))
108
109 machine.send_chars("mitmdump -s ${addonScript}\n")
110 machine.wait_for_open_port(8080)
111
112 t.assertEqual("success", curl("http://localhost:8000/old", proxy=True))
113
114 machine.send_key("ctrl-c")
115
116 # replay
117 # https://docs.mitmproxy.org/stable/mitmproxytutorial-replayrequests/
118 # https://docs.mitmproxy.org/stable/tutorials/client-replay/
119
120 t.assertEqual("0", curl("http://localhost:8000/counter"))
121
122 machine.send_chars("mitmdump -w replay\n")
123 machine.wait_for_open_port(8080)
124
125 curl("-X POST http://localhost:8000/counter", proxy=True)
126
127 machine.send_key("ctrl-c")
128
129 t.assertEqual("1", curl("http://localhost:8000/counter"))
130
131 machine.succeed("mitmdump -C /root/replay")
132
133 t.assertEqual("2", curl("http://localhost:8000/counter"))
134
135 # create a OpenAPI 3.0 spec from captured flow
136 # https://github.com/alufers/mitmproxy2swagger
137
138 # create a initial spec
139 machine.succeed("mitmproxy2swagger -i /root/replay -f flow -o /root/spec -p http://localhost:8000")
140 # don't ignore any endpoint
141 machine.succeed("sed -i -e 's/- ignore:/- /' /root/spec")
142 # generate the actual spec
143 machine.succeed("mitmproxy2swagger -i /root/replay -f flow -o /root/spec -p http://localhost:8000")
144 # check for endpoint /counter
145 machine.succeed("grep '/counter:' /root/spec")
146 '';
147}