107 lines
3.7 KiB
Python
107 lines
3.7 KiB
Python
"""
|
||
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' https://cdn.jsdelivr.net; "
|
||
"style-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net; "
|
||
"font-src 'self' https://cdn.jsdelivr.net; "
|
||
"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))
|
||
|
||
|
||
def validate_setting_key(key: str) -> bool:
|
||
"""Return True if key is a valid setting key (alphanumeric, underscores, dots)."""
|
||
import re
|
||
return bool(key and re.match(r'^[a-zA-Z0-9_.]+$', key) and len(key) <= 100)
|