05/06/2026 Initial commit
This commit is contained in:
@@ -0,0 +1,99 @@
|
||||
"""
|
||||
security.py — Security middleware and input sanitisation helpers.
|
||||
- apply_security_headers(app) — Attach security headers to every response.
|
||||
- check_admin_ip(app) — Enforce ADMIN_IP_ALLOWLIST at the Flask layer.
|
||||
- sanitise_string(value) — Strip control characters from user input.
|
||||
- validate_slug(slug) — Validate tenant slug format.
|
||||
"""
|
||||
|
||||
import re
|
||||
import logging
|
||||
import ipaddress
|
||||
from flask import request, abort, current_app
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Regex: slugs must be lowercase alphanumeric + hyphens, 2–80 chars
|
||||
_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9\-]{1,78}[a-z0-9]$")
|
||||
|
||||
# Strip ASCII control characters (0x00–0x1F, 0x7F) and Unicode C0/C1 blocks
|
||||
_CONTROL_CHAR_RE = re.compile(r"[\x00-\x1f\x7f\x80-\x9f]")
|
||||
|
||||
|
||||
def apply_security_headers(app):
|
||||
"""
|
||||
Register an after_request hook that attaches security headers to every
|
||||
response. Nginx adds HSTS and the more restrictive CSP for production;
|
||||
this layer provides defence-in-depth and covers the development server.
|
||||
"""
|
||||
@app.after_request
|
||||
def _add_headers(response):
|
||||
response.headers["X-Content-Type-Options"] = "nosniff"
|
||||
response.headers["X-Frame-Options"] = "SAMEORIGIN"
|
||||
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
|
||||
# Basic CSP — tightened further in Nginx for production
|
||||
response.headers["Content-Security-Policy"] = (
|
||||
"default-src 'self'; "
|
||||
"script-src 'self'; "
|
||||
"style-src 'self' 'unsafe-inline'; "
|
||||
"img-src 'self' data:; "
|
||||
"frame-ancestors 'none';"
|
||||
)
|
||||
return response
|
||||
|
||||
|
||||
def check_admin_ip(app):
|
||||
"""
|
||||
Register a before_request hook on the admin app that validates the
|
||||
client IP against ADMIN_IP_ALLOWLIST. Provides a Flask-layer double-check
|
||||
behind Nginx's allow/deny directives.
|
||||
"""
|
||||
@app.before_request
|
||||
def _check_ip():
|
||||
allowlist_raw = app.config.get("ADMIN_IP_ALLOWLIST", "")
|
||||
if not allowlist_raw.strip():
|
||||
return # No allowlist configured — skip check (dev mode)
|
||||
|
||||
networks = []
|
||||
for cidr in allowlist_raw.split(","):
|
||||
cidr = cidr.strip()
|
||||
if cidr:
|
||||
try:
|
||||
networks.append(ipaddress.ip_network(cidr, strict=False))
|
||||
except ValueError:
|
||||
logger.error("Invalid CIDR in ADMIN_IP_ALLOWLIST: %s", cidr)
|
||||
|
||||
if not networks:
|
||||
return
|
||||
|
||||
client_ip_str = request.headers.get("X-Real-IP") or request.remote_addr
|
||||
try:
|
||||
client_ip = ipaddress.ip_address(client_ip_str)
|
||||
except ValueError:
|
||||
logger.warning("Unparseable client IP: %s", client_ip_str)
|
||||
abort(403)
|
||||
return
|
||||
|
||||
if not any(client_ip in net for net in networks):
|
||||
logger.warning(
|
||||
"Admin IP blocked: %s not in allowlist", client_ip_str
|
||||
)
|
||||
abort(403)
|
||||
|
||||
|
||||
def sanitise_string(value: str, max_length: int = None) -> str:
|
||||
"""
|
||||
Strip control characters from a user-supplied string.
|
||||
Optionally truncate to max_length.
|
||||
"""
|
||||
if not isinstance(value, str):
|
||||
return value
|
||||
cleaned = _CONTROL_CHAR_RE.sub("", value).strip()
|
||||
if max_length:
|
||||
cleaned = cleaned[:max_length]
|
||||
return cleaned
|
||||
|
||||
|
||||
def validate_slug(slug: str) -> bool:
|
||||
"""Return True if the slug matches the allowed pattern."""
|
||||
return bool(_SLUG_RE.match(slug))
|
||||
Reference in New Issue
Block a user