pyrtan / sourcev0.14.0 · main

pyrtan / tests / test_access.py

11,192 bytes · last changed 2026-10-25 in Add tokens, CORS, persistent gaps, egress policy, the catalog and the console

1"""Tokens, CORS, persistent gaps and the egress policy: what an operator
2turns on when a swarm leaves the LAN or hosts trees for other people."""
3
4import json
5import socket
6import urllib.error
7import urllib.request
8
9import pytest
10
11from pyrtan import Registry, RecordedTransport
12from pyrtan.gaps import GapStore
13from pyrtan.hub import Hub
14from pyrtan.service import Server, Service, http_json
15from pyrtan.transport import EgressRefused, UrllibTransport, check_egress
16from conftest import EXAMPLES, FIXTURES
17
18GAPPY = """
19name = "gappy"
20description = "decides on a value it was given, with no else"
21start = "look"
22
23[[triggers]]
24pattern = "gappy {value:int}"
25
26[inputs.value]
27type = "int"
28
29[nodes.look]
30kind = "decide"
31on = "inputs.value"
32branches = [ { when = { eq = 1 }, then = "one" } ]
33
34[nodes.one]
35kind = "done"
36say = "one"
37"""
38
39
40def _raw(url: str, headers: dict | None = None, method: str = "GET", data: bytes | None = None):
41 req = urllib.request.Request(url, data=data, method=method, headers=headers or {})
42 try:
43 with urllib.request.urlopen(req) as resp:
44 return resp.status, dict(resp.headers), resp.read()
45 except urllib.error.HTTPError as e:
46 return e.code, dict(e.headers), e.read()
47
48
49@pytest.fixture
50def locked_service():
51 reg = Registry.from_paths([EXAMPLES])
52 svc = Service(reg, name="locked", transport=RecordedTransport.from_file(FIXTURES / "examples.json"), token="s3cret")
53 server = Server(svc.route, quiet=True, token="s3cret", cors=["https://pyrtan.com"]).start()
54 yield svc, server
55 server.stop()
56
57
58def test_token_required_except_public_paths(locked_service):
59 _, server = locked_service
60 status, headers, body = _raw(f"{server.url}/capabilities")
61 assert status == 401 and headers["WWW-Authenticate"].startswith("Bearer")
62 assert json.loads(body)["reason"] == "unauthorised"
63 assert _raw(f"{server.url}/intent", {"Content-Type": "application/json"}, "POST", b'{"text":"weather in Oslo"}')[0] == 401
64 assert _raw(f"{server.url}/capabilities", {"Authorization": "Bearer wrong"})[0] == 401
65 assert _raw(f"{server.url}/capabilities", {"Authorization": "Basic s3cret"})[0] == 401
66 status, _, body = _raw(f"{server.url}/capabilities", {"Authorization": "Bearer s3cret"})
67 assert status == 200 and {t["name"] for t in json.loads(body)["trees"]} >= {"weather"}
68 assert http_json("POST", f"{server.url}/intent", {"text": "weather in Oslo"}, token="s3cret")[1]["status"] == "done"
69 # Public: the console, its index, and liveness.
70 assert _raw(f"{server.url}/")[0] == 200
71 assert _raw(f"{server.url}/console")[0] == 200
72 status, _, body = _raw(f"{server.url}/api")
73 assert status == 200 and json.loads(body)["auth"] == "bearer"
74
75
76def test_health_is_trimmed_without_the_token(locked_service):
77 _, server = locked_service
78 status, _, body = _raw(f"{server.url}/health")
79 public = json.loads(body)
80 assert status == 200 and public["status"] == "ok" and "trees" not in public and "errors" not in public
81 status, _, body = _raw(f"{server.url}/health", {"Authorization": "Bearer s3cret"})
82 assert "weather" in json.loads(body)["trees"]
83
84
85def test_cors_only_for_listed_origins(locked_service):
86 _, server = locked_service
87 status, headers, _ = _raw(f"{server.url}/intent", {"Origin": "https://pyrtan.com", "Access-Control-Request-Method": "POST", "Access-Control-Request-Headers": "authorization,content-type"}, "OPTIONS")
88 assert status == 204
89 assert headers["Access-Control-Allow-Origin"] == "https://pyrtan.com"
90 assert "Authorization" in headers["Access-Control-Allow-Headers"]
91 status, headers, _ = _raw(f"{server.url}/intent", {"Origin": "https://evil.example"}, "OPTIONS")
92 assert status == 204 and "Access-Control-Allow-Origin" not in headers
93 _, headers, _ = _raw(f"{server.url}/api", {"Origin": "https://pyrtan.com"})
94 assert headers["Access-Control-Allow-Origin"] == "https://pyrtan.com" and headers["Vary"] == "Origin"
95 _, headers, _ = _raw(f"{server.url}/api")
96 assert "Access-Control-Allow-Origin" not in headers
97
98
99def test_cors_star_echoes_any_origin():
100 reg = Registry.from_paths([EXAMPLES])
101 server = Server(Service(reg).route, quiet=True, cors=["*"]).start()
102 try:
103 _, headers, _ = _raw(f"{server.url}/api", {"Origin": "http://localhost:5173"})
104 assert headers["Access-Control-Allow-Origin"] == "http://localhost:5173"
105 finally:
106 server.stop()
107
108
109def test_swarm_shares_one_token(tmp_path):
110 token = "swarm-token"
111 reg_a = Registry.from_paths([EXAMPLES])
112 svc_a = Service(reg_a, name="a", transport=RecordedTransport.from_file(FIXTURES / "examples.json"), token=token)
113 srv_a = Server(svc_a.route, quiet=True, token=token).start()
114 hub = Hub(poll_every=60, token=token)
115 hub_srv = Server(hub.route, quiet=True, token=token).start()
116 try:
117 # A stranger cannot join the swarm or use it.
118 assert http_json("POST", f"{hub_srv.url}/register", {"url": srv_a.url})[0] == 401
119 assert http_json("POST", f"{hub_srv.url}/intent", {"text": "weather in Oslo"})[0] == 401
120 # A service with the token registers; the hub polls it with the token.
121 svc_a.hub = hub_srv.url
122 assert svc_a.register(srv_a.url)
123 _, peers = http_json("GET", f"{hub_srv.url}/peers", token=token)
124 assert peers["peers"][0]["link"] == "up" and "weather" in peers["peers"][0]["trees"]
125 # And forwards with it.
126 _, out = http_json("POST", f"{hub_srv.url}/intent", {"text": "weather in Oslo"}, token=token)
127 assert out["status"] == "done" and out["peer"] == srv_a.url
128 # A hub without the token sees the peer as down.
129 other = Hub(poll_every=60)
130 p = other.add_peer(srv_a.url)
131 assert p.link == "down" and "401" in p.last_error
132 finally:
133 hub.stop()
134 hub_srv.stop()
135 srv_a.stop()
136
137
138# ---------------------------------------------------------------- gaps
139
140
141def test_gaps_persist_and_count(tmp_path):
142 (tmp_path / "gappy.toml").write_text(GAPPY)
143 path = tmp_path / "gaps.jsonl"
144 reg = Registry.from_paths([tmp_path])
145 svc = Service(reg, name="g", gaps=path)
146 server = Server(svc.route, quiet=True).start()
147 try:
148 assert http_json("POST", f"{server.url}/intent", {"text": "gappy 1"})[1]["status"] == "done"
149 for _ in range(3):
150 out = http_json("POST", f"{server.url}/intent", {"text": "gappy 7"})[1]
151 assert out["status"] == "error" and out["reason"] == "gap"
152 http_json("POST", f"{server.url}/intent", {"text": "gappy 8"})
153 _, gaps = http_json("GET", f"{server.url}/gaps")
154 assert [(g["node"], g["observed"], g["count"]) for g in gaps["gaps"]] == [("look", 7, 3), ("look", 8, 1)]
155 assert http_json("GET", f"{server.url}/health")[1]["gaps"] == 2
156 finally:
157 server.stop()
158 assert len(path.read_text().splitlines()) == 4
159 # A new service over the same file starts with what the old one saw.
160 again = GapStore(path)
161 assert [(g["observed"], g["count"]) for g in again.all()] == [(7, 3), (8, 1)]
162 again.add([{"tree": "gappy", "node": "look", "on": "inputs.value", "observed": 8}])
163 assert [(g["observed"], g["count"]) for g in again.all()] == [(7, 3), (8, 2)]
164
165
166def test_gap_store_keeps_going_when_the_file_cannot_be_written(tmp_path, capsys):
167 blocked = tmp_path / "dir"
168 blocked.mkdir()
169 store = GapStore(blocked) # a directory, not a file: reading and every append fail
170 store.add([{"tree": "t", "node": "n", "on": "x", "observed": 1}])
171 err = capsys.readouterr().err
172 assert len(store) == 1 and store.error and "cannot read gaps" in err
173 # A store over a file that cannot be written warns once and keeps the gap.
174 missing_dir = tmp_path / "nope"
175 missing_dir.write_text("") # a file where the parent directory should be
176 store = GapStore(missing_dir / "gaps.jsonl")
177 store.add([{"tree": "t", "node": "n", "on": "x", "observed": 1}] * 2)
178 assert store.all()[0]["count"] == 2 and store.error
179 assert capsys.readouterr().err.count("cannot write gaps") == 1
180
181
182def test_gap_store_in_memory_when_no_path():
183 store = GapStore()
184 store.add([{"tree": "t", "node": "n", "on": "x", "observed": "a"}] * 2)
185 assert len(store) == 1 and store.all()[0]["count"] == 2 and store.path is None
186
187
188# ---------------------------------------------------------------- egress
189
190
191def _resolver(*answers: str):
192 def resolve(host, port, proto=0):
193 return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (a, port)) for a in answers]
194
195 return resolve
196
197
198def test_egress_any_allows_everything_http():
199 check_egress("http://127.0.0.1:8701/health", "any")
200 check_egress("http://10.0.0.5/", "any", resolve=_resolver("10.0.0.5"))
201 with pytest.raises(EgressRefused):
202 check_egress("ftp://example.com/", "any")
203
204
205@pytest.mark.parametrize("address", ["127.0.0.1", "10.1.2.3", "172.16.0.9", "192.168.1.20", "169.254.169.254", "0.0.0.0", "::1", "fd00::1", "fe80::1", "::ffff:10.0.0.1", "224.0.0.1"])
206def test_egress_public_refuses_private_addresses(address):
207 with pytest.raises(EgressRefused):
208 check_egress("https://api.example.com/", "public", resolve=_resolver(address))
209
210
211def test_egress_public_allows_public_addresses():
212 check_egress("https://api.example.com/", "public", resolve=_resolver("93.184.216.34"))
213 check_egress("https://api.example.com/", "public", resolve=_resolver("2606:2800:220:1:248:1893:25c8:1946"))
214
215
216def test_egress_public_refuses_localhost_names_and_mixed_answers():
217 with pytest.raises(EgressRefused):
218 check_egress("http://localhost:8700/", "public")
219 with pytest.raises(EgressRefused):
220 check_egress("http://thing.localhost/", "public")
221 # One private answer among public ones is enough to refuse.
222 with pytest.raises(EgressRefused):
223 check_egress("https://api.example.com/", "public", resolve=_resolver("93.184.216.34", "10.0.0.1"))
224 with pytest.raises(EgressRefused):
225 check_egress("https://nope.invalid/", "public", resolve=lambda *a, **k: (_ for _ in ()).throw(socket.gaierror("no such host")))
226
227
228def test_egress_public_transport_refuses_before_connecting():
229 with pytest.raises(ValueError):
230 UrllibTransport(egress="strict")
231 with pytest.raises(EgressRefused):
232 UrllibTransport(egress="public").request("GET", "http://127.0.0.1:9/")
233
234
235def test_service_with_public_egress_reports_it_and_refuses_a_lan_host(tmp_path):
236 (tmp_path / "status.toml").write_text((EXAMPLES / "site-status" / "site-status.toml").read_text())
237 svc = Service(Registry.from_paths([tmp_path]), name="sealed", egress="public")
238 server = Server(svc.route, quiet=True).start()
239 try:
240 assert http_json("GET", f"{server.url}/health")[1]["egress"] == "public"
241 _, out = http_json("POST", f"{server.url}/run", {"tree": "site-status", "inputs": {"host": "127.0.0.1"}})
242 assert out["status"] == "error" and out["reason"] == "transport" and "not a public address" in out["say"]
243 # Refused once, not retried: one step in the trace.
244 assert [s["node"] for s in out["trace"]["steps"]] == ["probe"]
245 finally:
246 server.stop()