mirror of
https://github.com/Av0idD4rk/firegex.git
synced 2026-10-23 16:31:00 +00:00
159 lines
7.1 KiB
Python
159 lines
7.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Verifies the IP-based access control middleware (IPFilterMiddleware in backend/app.py),
|
|
configured via `run.py`'s --allowed-ips / --proxy-ip-header flags (ALLOWED_IPS / PROXY_IP_HEADER
|
|
env vars).
|
|
|
|
Unlike the rest of the test suite, this script manages firegex's own start/stop lifecycle,
|
|
since ALLOWED_IPS/PROXY_IP_HEADER are only read once at process startup (no live reload) -
|
|
each scenario needs a fresh restart with different flags. Must be run on the same Linux/Docker
|
|
host firegex runs on (same requirements as run.py itself).
|
|
"""
|
|
import argparse
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
import requests
|
|
|
|
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
BASE = "http://127.0.0.1:4444/"
|
|
|
|
failures = []
|
|
|
|
|
|
def check(name, cond, detail=""):
|
|
if cond:
|
|
print(f"\033[32mPASS: {name}\033[0m")
|
|
else:
|
|
print(f"\033[31mFAIL: {name}{' - ' + detail if detail else ''}\033[0m")
|
|
failures.append(name)
|
|
|
|
|
|
def run_py(*args, timeout=180):
|
|
res = subprocess.run(["python3", "run.py", *args], cwd=REPO_ROOT, timeout=timeout)
|
|
return res.returncode
|
|
|
|
|
|
def wait_ready(timeout=90):
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
try:
|
|
# A bare /api/status request is never blocked by design considerations of this
|
|
# test (it's the endpoint we probe), so poll it via a raw request expecting
|
|
# either 200 or 403 - both mean "the app answered", unlike a connection error.
|
|
r = requests.get(BASE + "api/status", timeout=2)
|
|
if r.status_code in (200, 403):
|
|
return True
|
|
except requests.RequestException:
|
|
pass
|
|
time.sleep(1)
|
|
return False
|
|
|
|
|
|
def restart_with(allowed_ips=None, proxy_ip_header=None, disable_auth=None):
|
|
print(f"\n--- restarting firegex (allowed_ips={allowed_ips!r}, proxy_ip_header={proxy_ip_header!r}, disable_auth={disable_auth!r}) ---")
|
|
run_py("stop")
|
|
args = ["start", "-P", "testpassword"]
|
|
# An empty string still needs to be passed explicitly to clear a previous value.
|
|
if allowed_ips is not None:
|
|
args += ["--allowed-ips", allowed_ips]
|
|
if proxy_ip_header is not None:
|
|
args += ["--proxy-ip-header", proxy_ip_header]
|
|
if disable_auth is not None:
|
|
args += ["--disable-auth" if disable_auth else "--no-disable-auth"]
|
|
run_py(*args)
|
|
if not wait_ready():
|
|
print("FATAL: firegex didn't come back up after restart")
|
|
sys.exit(1)
|
|
|
|
|
|
def raw_ws_handshake_rejected(path="/sock/socket.io/?EIO=4&transport=websocket", extra_headers=None, timeout=5):
|
|
"""Best-effort check: performs a raw WS upgrade handshake and reports whether the
|
|
server completed it (101 Switching Protocols) or refused/closed the connection.
|
|
Returns True if the handshake looks rejected, False if it looks accepted, None if
|
|
inconclusive (e.g. connection reset before any bytes came back)."""
|
|
lines = [
|
|
f"GET {path} HTTP/1.1",
|
|
"Host: 127.0.0.1:4444",
|
|
"Upgrade: websocket",
|
|
"Connection: Upgrade",
|
|
"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==",
|
|
"Sec-WebSocket-Version: 13",
|
|
]
|
|
for k, v in (extra_headers or {}).items():
|
|
lines.append(f"{k}: {v}")
|
|
req = "\r\n".join(lines) + "\r\n\r\n"
|
|
try:
|
|
with socket.create_connection(("127.0.0.1", 4444), timeout=timeout) as s:
|
|
s.sendall(req.encode())
|
|
data = s.recv(4096)
|
|
except (ConnectionResetError, OSError):
|
|
return True
|
|
if not data:
|
|
return True
|
|
return b"101" not in data.split(b"\r\n", 1)[0]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--skip-cleanup-restart", action="store_true",
|
|
help="Don't restart firegex without restrictions at the end (leaves the last scenario's config applied)")
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
# 1. No restriction configured -> unrestricted access (backward compatible default)
|
|
restart_with(allowed_ips="")
|
|
r = requests.get(BASE + "api/status")
|
|
check("no ALLOWED_IPS configured -> request succeeds", r.status_code == 200, f"got {r.status_code}")
|
|
|
|
# 2. Restrict to a network that excludes the loopback test client -> blocked
|
|
restart_with(allowed_ips="203.0.113.0/24")
|
|
r = requests.get(BASE + "api/status")
|
|
check("ALLOWED_IPS excludes client -> 403", r.status_code == 403, f"got {r.status_code}")
|
|
ws_rejected = raw_ws_handshake_rejected()
|
|
check("ALLOWED_IPS excludes client -> websocket upgrade rejected", ws_rejected is True, f"result={ws_rejected}")
|
|
|
|
# 3. Restrict to a network (CIDR) that includes the loopback test client -> allowed
|
|
restart_with(allowed_ips="127.0.0.1/32,::1/128")
|
|
r = requests.get(BASE + "api/status")
|
|
check("ALLOWED_IPS includes client (CIDR) -> 200", r.status_code == 200, f"got {r.status_code}")
|
|
|
|
# 4. PROXY_IP_HEADER: the header value is trusted over the raw TCP peer
|
|
restart_with(allowed_ips="203.0.113.55/32", proxy_ip_header="X-Forwarded-For")
|
|
r = requests.get(BASE + "api/status", headers={"X-Forwarded-For": "203.0.113.55"})
|
|
check("PROXY_IP_HEADER value matches allowlist -> 200", r.status_code == 200, f"got {r.status_code}")
|
|
r = requests.get(BASE + "api/status", headers={"X-Forwarded-For": "8.8.8.8"})
|
|
check("PROXY_IP_HEADER value NOT in allowlist -> 403", r.status_code == 403, f"got {r.status_code}")
|
|
r = requests.get(BASE + "api/status") # header absent -> falls back to raw peer (127.0.0.1, not allowed here)
|
|
check("PROXY_IP_HEADER configured but header absent -> falls back to raw peer -> 403",
|
|
r.status_code == 403, f"got {r.status_code}")
|
|
|
|
# 5. Malformed value in the trusted header must fail CLOSED (deny), not open -
|
|
# otherwise a client that can reach the port directly could send a garbage header
|
|
# value to bypass the allowlist entirely.
|
|
r = requests.get(BASE + "api/status", headers={"X-Forwarded-For": "not-an-ip"})
|
|
check("malformed PROXY_IP_HEADER value -> fails closed (403)", r.status_code == 403, f"got {r.status_code}")
|
|
|
|
# 6. Built-in authentication can be disabled for a reverse proxy that owns
|
|
# access control. Protected API routes and the status endpoint then work
|
|
# without a bearer token.
|
|
restart_with(allowed_ips="127.0.0.1/32", disable_auth=True)
|
|
r = requests.get(BASE + "api/status")
|
|
check("DISABLE_AUTH reports a logged-in running app", r.status_code == 200 and r.json().get("status") == "run" and r.json().get("loggined") is True, f"got {r.status_code}: {r.text}")
|
|
r = requests.get(BASE + "api/interfaces")
|
|
check("DISABLE_AUTH permits protected API without bearer token", r.status_code == 200, f"got {r.status_code}")
|
|
|
|
finally:
|
|
if not args.skip_cleanup_restart:
|
|
restart_with(allowed_ips="", disable_auth=False)
|
|
|
|
print()
|
|
if failures:
|
|
print(f"\033[31m{len(failures)} check(s) failed: {failures}\033[0m")
|
|
sys.exit(1)
|
|
else:
|
|
print("\033[32mAll IP filter checks passed\033[0m")
|