121 lines
5.5 KiB
Python
121 lines
5.5 KiB
Python
import json
|
|
from ipaddress import ip_address
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from starlette.requests import Request
|
|
|
|
import ip_access as access
|
|
|
|
|
|
@pytest.fixture
|
|
def policy(tmp_path, monkeypatch):
|
|
path = tmp_path / "access.json"
|
|
path.write_text(json.dumps({"allowed_ips": []}))
|
|
monkeypatch.setattr(access, "CONFIG_PATH", path)
|
|
monkeypatch.setattr(access, "local_addresses", lambda: {ip_address("192.168.10.2"), ip_address("::1")})
|
|
return path
|
|
|
|
|
|
def request(host, headers=None, port=50000):
|
|
return Request({"type": "http", "path": "/api/v1/hairline/generate-52", "scheme": "http",
|
|
"client": (host, port), "headers": [(k.encode(), v.encode()) for k, v in (headers or {}).items()]})
|
|
|
|
|
|
@pytest.mark.parametrize("host", ["127.0.0.1", "::1", "::ffff:127.0.0.1", "192.168.10.2"])
|
|
def test_only_actual_local_addresses_allowed_by_default(policy, host):
|
|
assert access.check_access(request(host))[0]
|
|
|
|
|
|
@pytest.mark.parametrize("host", ["203.0.113.9", "192.168.10.3", "0.0.0.0", "invalid"])
|
|
def test_other_addresses_denied(policy, host):
|
|
assert not access.check_access(request(host))[0]
|
|
|
|
|
|
def test_whitelist_ipv4_ipv6_cidr_and_hot_reload(policy):
|
|
policy.write_text(json.dumps({"allowed_ips": ["203.0.113.9", "198.51.100.0/24", "2001:db8::/64"]}))
|
|
for host in ["203.0.113.9", "::ffff:203.0.113.9", "198.51.100.12", "2001:db8::12"]:
|
|
assert access.check_access(request(host))[0]
|
|
policy.write_text('{"allowed_ips": []}')
|
|
assert not access.check_access(request("203.0.113.9"))[0]
|
|
|
|
|
|
def test_configured_public_whitelist_is_exact(policy):
|
|
policy.write_text('{"allowed_ips": ["49.232.51.171"]}')
|
|
assert access.check_access(request("49.232.51.171"))[0]
|
|
assert not access.check_access(request("49.232.51.172"))[0]
|
|
|
|
|
|
@pytest.mark.parametrize("header", ["x-forwarded-for", "x-real-ip", "forwarded"])
|
|
def test_forwarding_headers_cannot_override_socket_peer(policy, header):
|
|
assert not access.check_access(request("203.0.113.9", {header: "127.0.0.1"}))[0]
|
|
assert access.check_access(request("127.0.0.1", {header: "203.0.113.9"}))[0]
|
|
|
|
|
|
def test_gateway_is_allowed_independently_of_end_user_ip(policy):
|
|
policy.write_text('{"allowed_ips": ["49.232.51.171"]}')
|
|
assert access.check_access(request("49.232.51.171", {"x-forwarded-for": "203.0.113.9"}))[0]
|
|
assert not access.check_access(request("203.0.113.9", {"x-forwarded-for": "49.232.51.171"}))[0]
|
|
|
|
|
|
def test_rewritten_peer_or_invalid_policy_fail_closed(policy):
|
|
assert not access.check_access(request("127.0.0.1", port=0))[0]
|
|
policy.write_text('{"allowed_ips": ["not-an-ip"]}')
|
|
assert not access.check_access(request("127.0.0.1"))[0]
|
|
policy.write_text('invalid JSON')
|
|
assert not access.check_access(request("127.0.0.1"))[0]
|
|
|
|
|
|
def test_worker_guard_covers_all_interfaces_before_auth(policy, monkeypatch):
|
|
import app as worker
|
|
monkeypatch.setattr(worker, "ACCEPT_PASSWORDS", ["access-test"])
|
|
headers = {"X-Internal-Token": "access-test"}
|
|
with TestClient(worker.app, client=("203.0.113.9", 50000)) as client:
|
|
assert client.post('/api/v1/hairline/generate-52').status_code == 403
|
|
assert client.post('/api/v1/hairline/generate-52/', headers=headers).status_code == 403
|
|
assert client.post('/api/v1/debug/hairline-height-compare').status_code == 403
|
|
assert client.post('/api/v1/hairline/generate', headers=headers).status_code == 403
|
|
assert client.post('/api/v1/face/measure', headers=headers).status_code == 403
|
|
assert client.post('/api/v1/redraw', headers=headers).status_code == 403
|
|
assert client.get('/health').status_code == 403
|
|
assert client.get('/openapi.json').status_code == 403
|
|
policy.write_text('{"allowed_ips": ["203.0.113.9"]}')
|
|
assert client.post('/api/v1/hairline/generate-52', headers=headers).json()['code'] == 1004
|
|
assert client.post('/api/v1/hairline/generate-52').status_code == 401
|
|
|
|
|
|
def test_gateway_does_not_apply_worker_ip_policy(policy, monkeypatch):
|
|
import importlib
|
|
gateway = importlib.import_module('gateway.app')
|
|
calls = []
|
|
async def proxy(request, path):
|
|
calls.append(path)
|
|
return {'code': 0}
|
|
monkeypatch.setattr(gateway, '_proxy', proxy)
|
|
# Deliberately skip lifespan: no workers or configuration are needed.
|
|
client = TestClient(gateway.app, client=('203.0.113.9', 50000))
|
|
assert client.post('/api/v1/hairline/generate-52').status_code == 404
|
|
assert client.post('/api/v1/hairline/generate').json()['code'] == 0
|
|
assert client.post('/api/v1/hair/grow').json()['code'] == 0
|
|
assert client.get('/static/test_interface52.html').status_code == 200
|
|
assert calls == ['/api/v1/hairline/generate', '/api/v1/hair/grow']
|
|
|
|
|
|
@pytest.mark.parametrize(("path", "expected_code"), [
|
|
("/api/v1/hairline/generate-52", 1004),
|
|
("/api/v1/face/b-end-points", 1007),
|
|
])
|
|
def test_52_and_72_use_worker_transport_auth_not_client_bearer(policy, monkeypatch, path, expected_code):
|
|
import app as worker
|
|
|
|
policy.write_text('{"allowed_ips": ["49.232.51.171"]}')
|
|
monkeypatch.setattr(worker, "ACCEPT_PASSWORDS", ["access-test"])
|
|
client = TestClient(worker.app, client=("49.232.51.171", 50000))
|
|
try:
|
|
denied = client.post(path, headers={"Authorization": "Bearer client-access-token"})
|
|
assert denied.status_code == 401 and denied.json()["code"] == 1009
|
|
accepted = client.post(path, headers={"X-Internal-Token": "access-test"})
|
|
assert accepted.status_code == 200 and accepted.json()["code"] == expected_code
|
|
finally:
|
|
client.close()
|