Add security hardening and PostgreSQL 18 upgrade support
Add ALLOWED_HOSTS configuration to restrict trusted hosts in TrustedHostMiddleware. Enhance SSRF protection to block all private, loopback, link-local, multicast, reserved, and unspecified IP addresses using ipaddress module and DNS resolution checks. Add encrypt_value/decrypt_value aliases for encryption functions. Upgrade PostgreSQL from 16 to 18.4 in Docker Compose with updated data directory path (/var/lib/postgresql). Add security_opt no
This commit is contained in:
@@ -30,6 +30,7 @@ class Settings(BaseSettings):
|
||||
LOGIN_LOCKOUT_MINUTES: int = 15
|
||||
RATE_LIMIT_DEFAULT: int = 100
|
||||
RATE_LIMIT_LOGIN: int = 10
|
||||
ALLOWED_HOSTS: str = "localhost,127.0.0.1,testserver,nexadash-api"
|
||||
|
||||
# Mail
|
||||
SMTP_HOST: str = ""
|
||||
|
||||
@@ -14,6 +14,10 @@ from apps.api.src.models.base import Base
|
||||
from apps.api.src.security.rate_limit import limiter
|
||||
|
||||
|
||||
def _parse_csv_setting(value: str) -> list[str]:
|
||||
return [item.strip() for item in value.split(",") if item.strip()]
|
||||
|
||||
|
||||
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
|
||||
async def dispatch(self, request: Request, call_next):
|
||||
response = await call_next(request)
|
||||
@@ -60,7 +64,7 @@ def create_app() -> FastAPI:
|
||||
allow_headers=["*"],
|
||||
)
|
||||
app.add_middleware(SecurityHeadersMiddleware)
|
||||
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["*"])
|
||||
app.add_middleware(TrustedHostMiddleware, allowed_hosts=_parse_csv_setting(settings.ALLOWED_HOSTS))
|
||||
|
||||
app.include_router(api_router)
|
||||
|
||||
|
||||
@@ -23,6 +23,10 @@ def decrypt_secret(ciphertext: str) -> str:
|
||||
return _get_fernet().decrypt(ciphertext.encode()).decode()
|
||||
|
||||
|
||||
encrypt_value = encrypt_secret
|
||||
decrypt_value = decrypt_secret
|
||||
|
||||
|
||||
def hash_token(token: str) -> str:
|
||||
return hashlib.sha256(token.encode()).hexdigest()
|
||||
|
||||
|
||||
@@ -1,16 +1,28 @@
|
||||
import ipaddress
|
||||
import socket
|
||||
from urllib.parse import urlparse
|
||||
|
||||
|
||||
BLOCKED_SCHEMES = {"file", "gopher", "ftp", "dict", "ldap", "tftp"}
|
||||
BLOCKED_HOSTS = {
|
||||
"localhost",
|
||||
"127.0.0.1",
|
||||
"0.0.0.0",
|
||||
"::1",
|
||||
"169.254.169.254", # AWS metadata
|
||||
}
|
||||
|
||||
|
||||
def _is_blocked_ip(value: str) -> bool:
|
||||
try:
|
||||
ip = ipaddress.ip_address(value)
|
||||
except ValueError:
|
||||
return False
|
||||
return (
|
||||
ip.is_private
|
||||
or ip.is_loopback
|
||||
or ip.is_link_local
|
||||
or ip.is_multicast
|
||||
or ip.is_reserved
|
||||
or ip.is_unspecified
|
||||
)
|
||||
|
||||
|
||||
def is_safe_url(url: str) -> bool:
|
||||
try:
|
||||
parsed = urlparse(url)
|
||||
@@ -21,12 +33,17 @@ def is_safe_url(url: str) -> bool:
|
||||
hostname = parsed.hostname
|
||||
if not hostname:
|
||||
return False
|
||||
if hostname.lower() in BLOCKED_HOSTS:
|
||||
hostname = hostname.strip().lower().rstrip(".")
|
||||
if hostname in BLOCKED_HOSTS:
|
||||
return False
|
||||
# Block internal IP ranges
|
||||
parts = hostname.split(".")
|
||||
if len(parts) == 4 and all(p.isdigit() for p in parts):
|
||||
first = int(parts[0])
|
||||
if first in {10, 127} or (first == 192 and parts[1] == "168"):
|
||||
if _is_blocked_ip(hostname):
|
||||
return False
|
||||
try:
|
||||
resolved = socket.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)
|
||||
except socket.gaierror:
|
||||
return False
|
||||
for result in resolved:
|
||||
address = result[4][0]
|
||||
if _is_blocked_ip(address):
|
||||
return False
|
||||
return True
|
||||
|
||||
Reference in New Issue
Block a user