117 lines
5.0 KiB
Python
117 lines
5.0 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(policy):
|
|
import app as worker
|
|
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/').status_code == 403
|
|
assert client.post('/api/v1/debug/hairline-height-compare').status_code == 403
|
|
assert client.post('/api/v1/hairline/generate').status_code == 403
|
|
assert client.post('/api/v1/face/measure').status_code == 403
|
|
assert client.post('/api/v1/redraw').status_code == 403
|
|
assert client.post('/api/v1/hair/grow-22').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').json()['code'] == 1004
|
|
assert client.post('/api/v1/hair/grow-22').status_code == 422
|
|
|
|
|
|
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_accept_allowlisted_peer_without_token(policy, path, expected_code):
|
|
import app as worker
|
|
|
|
policy.write_text('{"allowed_ips": ["49.232.51.171"]}')
|
|
client = TestClient(worker.app, client=("49.232.51.171", 50000))
|
|
try:
|
|
accepted = client.post(path)
|
|
assert accepted.status_code == 200 and accepted.json()["code"] == expected_code
|
|
finally:
|
|
client.close()
|