Restrict local API to loopback clients

This commit is contained in:
Hamza Ayed
2026-10-03 14:35:31 +03:00
parent d57efbdca3
commit c1c4ba2ecf
6 changed files with 45 additions and 5 deletions
+5 -1
View File
@@ -9,4 +9,8 @@ from app import auth, database
def authenticated_client(app: FastAPI) -> TestClient:
database.ensure_user(database.LOCAL_USER_ID)
token, _ = auth.issue_session(database.LOCAL_USER_ID)
return TestClient(app, headers={"Authorization": f"Bearer {token}"})
return TestClient(
app,
headers={"Authorization": f"Bearer {token}"},
client=("127.0.0.1", 8000),
)
+13 -2
View File
@@ -20,7 +20,7 @@ from app.main import app, create_local_session
class AuthenticationTests(unittest.TestCase):
def setUp(self) -> None:
self.client = TestClient(app)
self.client = TestClient(app, client=("127.0.0.1", 8000))
self.created_users: list[str] = []
def tearDown(self) -> None:
@@ -321,9 +321,20 @@ class AuthenticationTests(unittest.TestCase):
self.assertEqual(second_own.json()["files"], ["two.md"])
def test_local_bootstrap_is_restricted_to_loopback_clients(self) -> None:
remote = self.client.post("/v1/auth/local-session", json={})
remote_client = TestClient(app, client=("192.0.2.10", 43210))
remote = remote_client.post("/v1/auth/local-session", json={})
self.assertEqual(remote.status_code, 403, remote.text)
def test_service_rejects_non_loopback_clients_even_for_public_endpoints(self) -> None:
remote_client = TestClient(app, client=("198.51.100.23", 51000))
blocked = remote_client.get("/openapi.json")
local = self.client.get("/openapi.json")
self.assertEqual(blocked.status_code, 403, blocked.text)
self.assertEqual(blocked.json()["error"]["code"], "loopback_only")
self.assertTrue(blocked.headers.get("x-request-id"))
self.assertEqual(local.status_code, 200, local.text)
request = Request(
{
"type": "http",
+6 -2
View File
@@ -186,10 +186,14 @@ class KnowledgeIndexTests(unittest.TestCase):
owner_token, _ = auth.issue_session(owner_id)
other_token, _ = auth.issue_session(other_id)
owner_client = TestClient(
app, headers={"Authorization": f"Bearer {owner_token}"}
app,
headers={"Authorization": f"Bearer {owner_token}"},
client=("127.0.0.1", 8000),
)
other_client = TestClient(
app, headers={"Authorization": f"Bearer {other_token}"}
app,
headers={"Authorization": f"Bearer {other_token}"},
client=("127.0.0.1", 8000),
)
with patch.dict(
os.environ,