feat: add subnet-based CORS validation support for VPN/Tailscale origins

- Add ipaddress module for subnet parsing (10.0.0.0/24 format)
- Implement subnet validation in CORS middleware
- Separate individual IPs from subnet definitions in EXTRA_ALLOWED_ORIGINS
- Custom SubnetAwareCORSMiddleware for dynamic origin validation
- Support both exact IP matches and subnet ranges
- Backward compatible with existing ALLOWED_ORIGINS list
This commit is contained in:
2026-04-21 15:17:29 +03:00
parent 8825118795
commit 983d6e4bb4
4 changed files with 114 additions and 17 deletions

View File

@@ -1,4 +1,5 @@
import os
from ipaddress import ip_address, ip_network, AddressValueError
from . import config_loader # This triggers the automatic environment loading
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
@@ -23,11 +24,14 @@ log.info("Database tables verified.")
app = FastAPI(title="TFM aInventory API", version="1.1.0")
log.info("TFM aInventory API process started.")
# [SECURITY FIX M-01] CORS Configuration
# [SECURITY FIX M-01] CORS Configuration with Subnet Support
# We dynamically build allowed origins from environment variables to simplify deployment.
_raw_origins = os.environ.get("ALLOWED_ORIGINS", "")
ALLOWED_ORIGINS = [o.strip() for o in _raw_origins.split(",") if o.strip()]
# Allowed subnets for subnet-based CORS validation (e.g., VPN, Tailscale)
ALLOWED_SUBNETS = []
# Automatically add origins based on network_config.env variables if present
server_ip = os.environ.get("SERVER_IP")
front_port = os.environ.get("FRONTEND_PORT", "8917")
@@ -55,27 +59,108 @@ if server_ip and server_ip != "localhost":
if ip_o not in ALLOWED_ORIGINS:
ALLOWED_ORIGINS.append(ip_o)
# [NEW] Add Extra Allowed Origins (Tailscale, VPN, etc.)
# [NEW] Add Extra Allowed Origins (Tailscale, VPN, etc.) with Subnet Support
extra_origins_raw = os.environ.get("EXTRA_ALLOWED_ORIGINS", "")
if extra_origins_raw:
for extra_ip in [o.strip() for o in extra_origins_raw.split(",") if o.strip()]:
# Generate standard combinations for this extra origin
ext_combos = [
f"http://{extra_ip}:{front_port}",
f"https://{extra_ip}:{front_ssl_port}",
f"https://{extra_ip}:{back_ssl_port}",
]
for combo in ext_combos:
if combo not in ALLOWED_ORIGINS:
ALLOWED_ORIGINS.append(combo)
for extra_item in [o.strip() for o in extra_origins_raw.split(",") if o.strip()]:
# Check if it's a subnet (contains /) or individual IP
if "/" in extra_item:
try:
# Parse as subnet
subnet = ip_network(extra_item, strict=False)
ALLOWED_SUBNETS.append(subnet)
log.info(f" -> Subnet allowed: {extra_item}")
except (AddressValueError, ValueError) as e:
log.warning(f" ⚠️ Invalid subnet {extra_item}: {e}")
else:
# Treat as individual IP - generate standard port combinations
ext_combos = [
f"http://{extra_item}:{front_port}",
f"https://{extra_item}:{front_ssl_port}",
f"https://{extra_item}:{back_ssl_port}",
]
for combo in ext_combos:
if combo not in ALLOWED_ORIGINS:
ALLOWED_ORIGINS.append(combo)
log.info("🔒 [SECURITY] CORS configuration initialized.")
log.info(f" Exact origins: {len(ALLOWED_ORIGINS)}")
for origin in ALLOWED_ORIGINS:
log.info(f" -> Allowed: {origin}")
log.info(f" -> {origin}")
if ALLOWED_SUBNETS:
log.info(f" Allowed subnets: {len(ALLOWED_SUBNETS)}")
for subnet in ALLOWED_SUBNETS:
log.info(f" -> {subnet}")
# Helper function to check if origin is allowed (exact match or subnet)
def is_origin_allowed(origin: str) -> bool:
"""Check if origin is in allowed origins or matches any allowed subnet"""
# Check exact match first (faster)
if origin in ALLOWED_ORIGINS:
return True
# Check subnet match if subnets are configured
if not ALLOWED_SUBNETS:
return False
try:
# Extract IP from origin URL (e.g., "https://192.168.1.100:8919" -> "192.168.1.100")
from urllib.parse import urlparse
parsed = urlparse(origin)
origin_host = parsed.hostname
if not origin_host:
return False
origin_ip = ip_address(origin_host)
for subnet in ALLOWED_SUBNETS:
if origin_ip in subnet:
return True
except (AddressValueError, ValueError):
pass
return False
# Custom CORS middleware class that supports subnets
from starlette.middleware.cors import CORSMiddleware as StarletteCORSMiddleware
from starlette.types import ASGIApp, Receive, Scope, Send, Message
class SubnetAwareCORSMiddleware(StarletteCORSMiddleware):
def __init__(self, *args, **kwargs):
# Keep simple origins list for backward compatibility
super().__init__(*args, **kwargs)
def allow_all_origins(self) -> bool:
return False
def preflight_allowed_origin(self, origin: str) -> bool:
# Check using our custom function
return is_origin_allowed(origin)
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
# Override to use our subnet-aware checking
if scope["type"] != "http":
await self.app(scope, receive, send)
return
method = scope["method"]
headers = dict(scope["headers"])
origin = headers.get(b"origin", b"").decode()
if self.preflight_allowed_origin(origin):
scope["headers"] = [
(name, value) for (name, value) in scope["headers"]
if name.lower() != b"origin"
]
scope["headers"] += [
(b"origin", origin.encode()),
]
await super().__call__(scope, receive, send)
# Add CORS middleware FIRST (before rate limiter)
# Use custom middleware that supports subnet validation
app.add_middleware(
CORSMiddleware,
SubnetAwareCORSMiddleware if ALLOWED_SUBNETS else CORSMiddleware,
allow_origins=ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],