feat(link): device registry, immediate sign-out, and an audit trail
Answers three questions that had no answer: which computers are connected, how do I cut one off, and what is recorded. - state.json gains a device registry hanging off the identity that owns it, so 'which computers can reach my files' cannot drift from the identity map. Clients older than v1.2 send no device_id and fall back to the token name, so they still register. - POST /v1/devices lists them; POST /v1/devices/revoke deletes that device's forge token via admin basic auth + Sudo (verified 204 on 1.27.2, after which the token is 401 immediately). Revocation is deliberately NOT rate limited -- nobody should be throttled out of signing out a lost laptop. - The device id is now part of the token NAME. Revocation deletes by name, so two machines called 'macbook' linked in the same second would otherwise collide and signing one out would kill the other. - A failed forge deletion is not recorded as revoked: a registry claiming 'revoked' while the token still works is worse than an honest error. - Append-only JSONL audit log (0600, rotates at 64MB), separate from state.json because state is rewritten atomically on every change and an audit trail the audited thing can rewrite is not one. A failed audit write is logged loudly and never breaks the request. - Authorisation everywhere: the forge decides who a token belongs to (GET /api/v1/user). No login is ever read from the request body. - Client: devices / logout / activity. The audit log records granthi-link events only -- git pushes and pulls never pass through this service. /v1/audit returns that caveat in its own response rather than letting the log read as file activity. 172 tests (was 153).
This commit is contained in:
claude
+253
-5
@@ -9,6 +9,7 @@ import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from unittest import mock
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
@@ -43,6 +44,13 @@ class StubUpstream(BaseHTTPRequestHandler):
|
||||
"[email protected]", "email":
|
||||
"[email protected]", "name": "Alice"})
|
||||
return self._json(401, {"error": "invalid token"})
|
||||
if self.path == "/api/v1/user":
|
||||
# whoami: the forge decides who a token belongs to
|
||||
tok = self.headers.get("Authorization", "").replace("token ", "")
|
||||
login = st["tokens_by_sha"].get(tok)
|
||||
if not login:
|
||||
return self._json(401, {"message": "unauthorized"})
|
||||
return self._json(200, {"login": login})
|
||||
if self.path.startswith("/api/v1/users/") and not self.path.endswith("/tokens"):
|
||||
login = self.path.rsplit("/", 1)[1]
|
||||
if login in st["hide_once"]:
|
||||
@@ -71,7 +79,14 @@ class StubUpstream(BaseHTTPRequestHandler):
|
||||
"sudo": self.headers.get("Sudo", ""), "body": body})
|
||||
if not self.headers.get("Authorization", "").startswith("Basic "):
|
||||
return self._json(401, {"message": "auth required"})
|
||||
return self._json(201, {"sha1": "MINTED", "name": body["name"]})
|
||||
login = self.path.split("/")[4]
|
||||
# unique per mint, like a real forge -- two users may legitimately
|
||||
# hold the same token NAME
|
||||
st["sha_seq"] = st.get("sha_seq", 0) + 1
|
||||
sha = f"SHA-{login}-{st['sha_seq']}-{body['name']}"
|
||||
st["tokens_by_sha"][sha] = login
|
||||
st["tokens"].setdefault(login, set()).add(body["name"])
|
||||
return self._json(201, {"sha1": sha, "name": body["name"]})
|
||||
if self.path == "/api/v1/user/repos":
|
||||
if body["name"] in st["repos"]:
|
||||
return self._json(409, {"message": "exists"})
|
||||
@@ -81,6 +96,26 @@ class StubUpstream(BaseHTTPRequestHandler):
|
||||
f"alice/{body['name']}"})
|
||||
self._json(404, {})
|
||||
|
||||
def do_DELETE(self):
|
||||
st = self.state
|
||||
if self.path.startswith("/api/v1/users/") and "/tokens/" in self.path:
|
||||
if not self.headers.get("Authorization", "").startswith("Basic "):
|
||||
# matches real Gitea: token auth cannot delete tokens
|
||||
return self._json(403, {"message": "basic auth required"})
|
||||
login, name = self.path.split("/tokens/")
|
||||
login = login.rsplit("/", 1)[1]
|
||||
name = urllib.parse.unquote(name)
|
||||
if st.get("revoke_fails"):
|
||||
return self._json(500, {"message": "forge exploded"})
|
||||
if name not in st["tokens"].get(login, set()):
|
||||
return self._json(404, {"message": "not found"})
|
||||
st["tokens"][login].discard(name)
|
||||
for sha, owner in list(st["tokens_by_sha"].items()):
|
||||
if owner == login and sha.endswith(f"-{name}"):
|
||||
del st["tokens_by_sha"][sha]
|
||||
return self._json(204, {})
|
||||
self._json(404, {})
|
||||
|
||||
def log_message(self, *a):
|
||||
pass
|
||||
|
||||
@@ -88,7 +123,9 @@ class StubUpstream(BaseHTTPRequestHandler):
|
||||
class ServiceTestBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
StubUpstream.state = {"users": {}, "created": [], "repos": set(),
|
||||
"token_reqs": [], "hide_once": set()}
|
||||
"token_reqs": [], "hide_once": set(),
|
||||
"tokens": {}, "tokens_by_sha": {},
|
||||
"revoke_fails": False}
|
||||
self.upstream = ThreadingHTTPServer(("127.0.0.1", 0), StubUpstream)
|
||||
threading.Thread(target=self.upstream.serve_forever, daemon=True).start()
|
||||
self.addCleanup(self.upstream.shutdown)
|
||||
@@ -102,6 +139,7 @@ class ServiceTestBase(unittest.TestCase):
|
||||
"admin_token": "ADMTOK", "admin_login": "root",
|
||||
"admin_password": "rootpw", "test_mode": False,
|
||||
"state_path": self.state_path,
|
||||
"audit_path": os.path.join(self.state_dir, "audit.jsonl"),
|
||||
})
|
||||
|
||||
def enable_test_mode(self):
|
||||
@@ -111,13 +149,17 @@ class ServiceTestBase(unittest.TestCase):
|
||||
patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
|
||||
def stub_link(self, sub, username, email=None, verified=None, device="d"):
|
||||
def stub_link(self, sub, username, email=None, verified=None, device="d",
|
||||
device_id=None, client_ip=None):
|
||||
ui = {"sub": sub, "preferred_username": username}
|
||||
if email is not None:
|
||||
ui["email"] = email
|
||||
if verified is not None:
|
||||
ui["email_verified"] = verified
|
||||
return self.svc.link({"test_userinfo": ui, "device_name": device})
|
||||
body = {"test_userinfo": ui, "device_name": device}
|
||||
if device_id:
|
||||
body["device_id"] = device_id
|
||||
return self.svc.link(body, client_ip)
|
||||
|
||||
def read_state(self):
|
||||
with open(self.state_path) as f:
|
||||
@@ -146,7 +188,8 @@ class TestLink(ServiceTestBase):
|
||||
"device_name": "mac studio"})
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(resp["login"], "alice.smith")
|
||||
self.assertEqual(resp["token"], "MINTED")
|
||||
# the stub now issues a per-token sha so revocation can be tested
|
||||
self.assertIn("granthi-sync-", resp["token"])
|
||||
self.assertEqual(resp["gitea_base"], "http://public.example:3041")
|
||||
st = StubUpstream.state
|
||||
self.assertEqual(len(st["created"]), 1)
|
||||
@@ -480,6 +523,211 @@ class TestHealth(HandlerTestBase):
|
||||
self.assertEqual(body["service"], "granthi-link")
|
||||
|
||||
|
||||
|
||||
class TestDeviceRegistry(ServiceTestBase):
|
||||
"""Which computers can reach my files, and can I cut one off."""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.enable_test_mode()
|
||||
|
||||
def link_device(self, device_id, name="laptop"):
|
||||
status, resp = self.stub_link("s1", "alice", device=name,
|
||||
device_id=device_id)
|
||||
self.assertEqual(status, 200, resp)
|
||||
return resp
|
||||
|
||||
def test_devices_are_recorded_and_listed_per_account(self):
|
||||
self.link_device("dev-a", "work-laptop")
|
||||
second = self.link_device("dev-b", "home-mac")
|
||||
status, resp = self.svc.devices({"token": second["token"]})
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(resp["login"], "alice")
|
||||
ids = [d["device_id"] for d in resp["devices"]]
|
||||
self.assertEqual(sorted(ids), ["dev-a", "dev-b"])
|
||||
names = {d["device_id"]: d["name"] for d in resp["devices"]}
|
||||
self.assertEqual(names["dev-a"], "work-laptop")
|
||||
|
||||
def test_listing_needs_a_working_token(self):
|
||||
self.link_device("dev-a")
|
||||
status, _ = self.svc.devices({"token": "not-a-real-token"})
|
||||
self.assertEqual(status, 401)
|
||||
status, _ = self.svc.devices({})
|
||||
self.assertEqual(status, 401)
|
||||
|
||||
def test_login_is_never_taken_from_the_request_body(self):
|
||||
"""Authorisation comes from the forge's answer about the token, so a
|
||||
body claiming another account changes nothing."""
|
||||
first = self.link_device("dev-a")
|
||||
status, resp = self.svc.devices({"token": first["token"],
|
||||
"login": "somebody-else"})
|
||||
self.assertEqual(resp["login"], "alice")
|
||||
|
||||
def test_relinking_the_same_device_does_not_duplicate_it(self):
|
||||
self.link_device("dev-a", "laptop")
|
||||
self.link_device("dev-a", "laptop-renamed")
|
||||
status, resp = self.svc.devices({"token": self.link_device("dev-a")["token"]})
|
||||
self.assertEqual(len(resp["devices"]), 1)
|
||||
|
||||
def test_old_clients_without_a_device_id_still_register(self):
|
||||
resp = self.stub_link("s1", "alice", device="ancient")[1]
|
||||
self.assertEqual(resp["device_id"], resp["token_name"])
|
||||
status, listed = self.svc.devices({"token": resp["token"]})
|
||||
self.assertEqual([d["device_id"] for d in listed["devices"]],
|
||||
[resp["token_name"]])
|
||||
|
||||
|
||||
class TestRevoke(ServiceTestBase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.enable_test_mode()
|
||||
self.a = self.stub_link("s1", "alice", device="laptop-a",
|
||||
device_id="dev-a")[1]
|
||||
self.b = self.stub_link("s1", "alice", device="laptop-b",
|
||||
device_id="dev-b")[1]
|
||||
|
||||
def test_revoking_a_device_kills_its_token_at_the_forge(self):
|
||||
status, resp = self.svc.revoke_device({"token": self.b["token"],
|
||||
"device_id": "dev-a"})
|
||||
self.assertEqual(status, 200, resp)
|
||||
# the revoked device's token no longer authenticates ANYTHING
|
||||
self.assertEqual(self.svc.whoami(self.a["token"]), None)
|
||||
# the device that did the revoking still works
|
||||
self.assertEqual(self.svc.whoami(self.b["token"]), "alice")
|
||||
|
||||
def test_a_device_can_revoke_itself(self):
|
||||
status, _ = self.svc.revoke_device({"token": self.a["token"],
|
||||
"device_id": "dev-a"})
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(self.svc.whoami(self.a["token"]), None)
|
||||
|
||||
def test_revoked_device_is_marked_not_deleted(self):
|
||||
self.svc.revoke_device({"token": self.b["token"],
|
||||
"device_id": "dev-a"})
|
||||
status, resp = self.svc.devices({"token": self.b["token"]})
|
||||
by_id = {d["device_id"]: d for d in resp["devices"]}
|
||||
self.assertIsNotNone(by_id["dev-a"]["revoked_at"])
|
||||
self.assertIsNone(by_id["dev-b"]["revoked_at"])
|
||||
|
||||
def test_cannot_revoke_a_device_on_another_account(self):
|
||||
other = self.stub_link("s2", "mallory", device="theirs",
|
||||
device_id="dev-x")[1]
|
||||
status, resp = self.svc.revoke_device({"token": other["token"],
|
||||
"device_id": "dev-a"})
|
||||
self.assertEqual(status, 404)
|
||||
# alice's device is untouched
|
||||
self.assertEqual(self.svc.whoami(self.a["token"]), "alice")
|
||||
|
||||
def test_a_failed_forge_delete_is_not_recorded_as_revoked(self):
|
||||
"""A registry that says 'revoked' while the token still works is
|
||||
worse than an honest error."""
|
||||
StubUpstream.state["revoke_fails"] = True
|
||||
status, _ = self.svc.revoke_device({"token": self.b["token"],
|
||||
"device_id": "dev-a"})
|
||||
self.assertEqual(status, 502)
|
||||
StubUpstream.state["revoke_fails"] = False
|
||||
_, listed = self.svc.devices({"token": self.b["token"]})
|
||||
by_id = {d["device_id"]: d for d in listed["devices"]}
|
||||
self.assertIsNone(by_id["dev-a"]["revoked_at"])
|
||||
self.assertEqual(self.svc.whoami(self.a["token"]), "alice")
|
||||
|
||||
def test_revoking_twice_is_not_an_error(self):
|
||||
self.svc.revoke_device({"token": self.b["token"], "device_id": "dev-a"})
|
||||
status, _ = self.svc.revoke_device({"token": self.b["token"],
|
||||
"device_id": "dev-a"})
|
||||
self.assertEqual(status, 200) # forge 404 = already gone = success
|
||||
|
||||
|
||||
class TestAuditLog(ServiceTestBase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.enable_test_mode()
|
||||
|
||||
def test_link_and_revoke_are_recorded_with_who_and_where(self):
|
||||
first = self.stub_link("s1", "alice", device="laptop",
|
||||
device_id="dev-a", client_ip="203.0.113.9")[1]
|
||||
self.stub_link("s1", "alice", device="mac", device_id="dev-b")
|
||||
second = self.svc.devices({"token": first["token"]})[1]
|
||||
self.assertEqual(len(second["devices"]), 2)
|
||||
self.svc.revoke_device({"token": first["token"],
|
||||
"device_id": "dev-b"}, client_ip="198.51.100.4")
|
||||
|
||||
status, resp = self.svc.audit_read({"token": first["token"]})
|
||||
self.assertEqual(status, 200)
|
||||
events = resp["events"]
|
||||
kinds = [e["event"] for e in events]
|
||||
self.assertIn("device.link", kinds)
|
||||
self.assertIn("device.revoke", kinds)
|
||||
link_ev = [e for e in events if e["event"] == "device.link"
|
||||
and e["device_id"] == "dev-a"][0]
|
||||
self.assertEqual(link_ev["client_ip"], "203.0.113.9")
|
||||
self.assertEqual(link_ev["login"], "alice")
|
||||
revoke_ev = [e for e in events if e["event"] == "device.revoke"][0]
|
||||
self.assertEqual(revoke_ev["client_ip"], "198.51.100.4")
|
||||
self.assertEqual(revoke_ev["device_id"], "dev-b")
|
||||
|
||||
def test_events_are_newest_first_and_scoped_to_the_caller(self):
|
||||
mine = self.stub_link("s1", "alice", device_id="dev-a")[1]
|
||||
self.stub_link("s2", "mallory", device_id="dev-x")
|
||||
_, resp = self.svc.audit_read({"token": mine["token"]})
|
||||
self.assertTrue(resp["events"])
|
||||
self.assertTrue(all(e["login"] == "alice" for e in resp["events"]))
|
||||
stamps = [e["ts"] for e in resp["events"]]
|
||||
self.assertEqual(stamps, sorted(stamps, reverse=True))
|
||||
|
||||
def test_reading_the_log_needs_a_working_token(self):
|
||||
self.stub_link("s1", "alice", device_id="dev-a")
|
||||
status, _ = self.svc.audit_read({"token": "nope"})
|
||||
self.assertEqual(status, 401)
|
||||
|
||||
def test_a_refused_revoke_is_recorded_too(self):
|
||||
self.stub_link("s1", "alice", device_id="dev-a")
|
||||
other = self.stub_link("s2", "mallory", device_id="dev-x")[1]
|
||||
self.svc.revoke_device({"token": other["token"],
|
||||
"device_id": "dev-a"}, client_ip="192.0.2.7")
|
||||
_, resp = self.svc.audit_read({"token": other["token"]})
|
||||
denied = [e for e in resp["events"]
|
||||
if e["event"] == "device.revoke.denied"]
|
||||
self.assertEqual(len(denied), 1)
|
||||
self.assertEqual(denied[0]["client_ip"], "192.0.2.7")
|
||||
|
||||
def test_the_log_says_what_it_cannot_see(self):
|
||||
"""Reading this and believing it lists file activity would be a real
|
||||
mistake: git traffic never passes through this service."""
|
||||
mine = self.stub_link("s1", "alice", device_id="dev-a")[1]
|
||||
_, resp = self.svc.audit_read({"token": mine["token"]})
|
||||
self.assertIn("pushes and pulls", resp["note"])
|
||||
|
||||
def test_a_broken_audit_file_does_not_break_the_request(self):
|
||||
self.svc.audit.path = "/nonexistent-dir/audit.jsonl"
|
||||
status, _ = self.stub_link("s1", "alice", device_id="dev-a")
|
||||
self.assertEqual(status, 200) # linking still works
|
||||
|
||||
def test_rotation_keeps_the_previous_file_readable(self):
|
||||
self.svc.audit.keep_bytes = 1500 # one rotation, not many
|
||||
mine = self.stub_link("s1", "alice", device_id="dev-a")[1]
|
||||
for i in range(20):
|
||||
self.svc.audit.write("noise", login="alice", n=i)
|
||||
_, resp = self.svc.audit_read({"token": mine["token"]}, )
|
||||
self.assertGreater(len(resp["events"]), 15) # spans both files
|
||||
self.assertTrue(os.path.exists(self.svc.audit.path + ".1"))
|
||||
|
||||
|
||||
class TestTokenNameUniqueness(ServiceTestBase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.enable_test_mode()
|
||||
|
||||
def test_two_devices_named_the_same_get_different_token_names(self):
|
||||
"""Revocation deletes by token name, so a collision would mean
|
||||
signing out one laptop kills the other."""
|
||||
a = self.stub_link("s1", "alice", device="macbook", device_id="dev-a")[1]
|
||||
b = self.stub_link("s1", "alice", device="macbook", device_id="dev-b")[1]
|
||||
self.assertNotEqual(a["token_name"], b["token_name"])
|
||||
self.svc.revoke_device({"token": b["token"], "device_id": "dev-a"})
|
||||
self.assertIsNone(self.svc.whoami(a["token"]))
|
||||
self.assertEqual(self.svc.whoami(b["token"]), "alice")
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user