192 lines
7.6 KiB
Python
192 lines
7.6 KiB
Python
import re
|
||
import uuid
|
||
import logging
|
||
from gvsds.app_config import getlist as _app_getlist, get as _app_get, getboolean as _app_getbool
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class DomainMiddleware:
|
||
def __init__(self, get_response):
|
||
self.get_response = get_response
|
||
self.known_domains = set(_app_getlist('domain', 'KNOWN_DOMAINS', fallback=[
|
||
'gvsds.com', 'gvsds.dev', 'yc8z.org', 'ycbz.org', 'gvsds.org', 'gvsds.studio',
|
||
]))
|
||
self._ip_pattern = re.compile(r'^\d{1,3}(\.\d{1,3}){3}$')
|
||
|
||
def __call__(self, request):
|
||
host = request.get_host().split(':')[0]
|
||
request.subdomain = 'www'
|
||
|
||
if self._ip_pattern.match(host):
|
||
pass
|
||
elif host.startswith('[') or ':' in host:
|
||
pass
|
||
else:
|
||
domain_parts = host.split('.')
|
||
if len(domain_parts) >= 3:
|
||
request.subdomain = domain_parts[0]
|
||
elif len(domain_parts) == 2:
|
||
if domain_parts[1] in ('localhost', '127.0.0.1'):
|
||
request.subdomain = domain_parts[0]
|
||
elif domain_parts[0] in self.known_domains or domain_parts[1] in self.known_domains:
|
||
request.subdomain = 'www'
|
||
|
||
response = self.get_response(request)
|
||
|
||
# CORS 由 django-cors-headers (CorsMiddleware) 统一处理，此处不再手动设置
|
||
|
||
csp_domain = _app_get('domain', 'CSP_DOMAIN', fallback='gvsds.com')
|
||
allowed_frame_ancestors = f"'self' https://www.{csp_domain} https://gerp.{csp_domain} https://ugm.{csp_domain}"
|
||
response['Content-Security-Policy'] = f"frame-ancestors {allowed_frame_ancestors}"
|
||
|
||
return response
|
||
|
||
def process_exception(self, request, exception):
|
||
return None
|
||
|
||
|
||
from gvsdsdk.auth.middleware import CurrentRoleMiddleware
|
||
|
||
|
||
class ErrorLoggingMiddleware:
|
||
def __init__(self, get_response):
|
||
self.get_response = get_response
|
||
|
||
def __call__(self, request):
|
||
response = self.get_response(request)
|
||
if getattr(response, 'status_code', 0) >= 500:
|
||
exc = getattr(request, '_caught_exception', None)
|
||
if exc:
|
||
logger.error(
|
||
'[%s] [%d] [%s] [%s] %s',
|
||
request.META.get('HTTP_HOST', 'unknown'),
|
||
response.status_code,
|
||
request.method,
|
||
request.META.get('REMOTE_ADDR', '-'),
|
||
request.get_full_path(),
|
||
exc_info=(type(exc), exc, exc.__traceback__),
|
||
)
|
||
else:
|
||
logger.error(
|
||
'[%s] [%d] [%s] [%s] %s',
|
||
request.META.get('HTTP_HOST', 'unknown'),
|
||
response.status_code,
|
||
request.method,
|
||
request.META.get('REMOTE_ADDR', '-'),
|
||
request.get_full_path(),
|
||
)
|
||
return response
|
||
|
||
def process_exception(self, request, exception):
|
||
request._caught_exception = exception
|
||
logger.error(
|
||
'[%s] [500] [%s] [%s] %s',
|
||
request.META.get('HTTP_HOST', 'unknown'),
|
||
request.method,
|
||
request.META.get('REMOTE_ADDR', '-'),
|
||
request.get_full_path(),
|
||
exc_info=(type(exception), exception, exception.__traceback__),
|
||
)
|
||
return None
|
||
|
||
|
||
class AttackDetectMiddleware:
|
||
"""攻击检测中间件——记录 401/403/429 响应到 AbnormalUserLog
|
||
|
||
自动捕获鉴权失败、越权访问、限流触发等可疑请求，记录完整请求指纹
|
||
（IP/UA/方法/路径/参数/请求头摘要）用于安全审计与攻击溯源。
|
||
|
||
配置（app.ini [attack_detect] 段）：
|
||
ENABLED = true/false（默认 true）
|
||
STATUS_CODES = 401,403,429（触发记录的状态码）
|
||
EXEMPT_PATHS = /API/Auth,/health（免记录路径）
|
||
MAX_BODY_LEN = 1024（请求体摘要最大长度）
|
||
"""
|
||
|
||
def __init__(self, get_response):
|
||
self.get_response = get_response
|
||
self.enabled = _app_getbool('attack_detect', 'ENABLED', fallback=True)
|
||
self.status_codes = set(
|
||
int(s) for s in _app_getlist('attack_detect', 'STATUS_CODES', fallback=['401', '403', '429'])
|
||
)
|
||
self.exempt_paths = tuple(
|
||
p.strip() for p in _app_getlist('attack_detect', 'EXEMPT_PATHS', fallback=['/API/Auth', '/health'])
|
||
)
|
||
self.max_body_len = int(_app_get('attack_detect', 'MAX_BODY_LEN', fallback='1024'))
|
||
|
||
def __call__(self, request):
|
||
response = self.get_response(request)
|
||
if not self.enabled:
|
||
return response
|
||
if response.status_code not in self.status_codes:
|
||
return response
|
||
path = request.path
|
||
if path.startswith(self.exempt_paths):
|
||
return response
|
||
self._log_attack(request, response)
|
||
return response
|
||
|
||
def _log_attack(self, request, response):
|
||
"""记录攻击日志到 AbnormalUserLog。"""
|
||
try:
|
||
from gvsds.apps.gerp.models import AbnormalUserLog
|
||
from gvsdsdk.auth.middleware import CurrentRoleMiddleware
|
||
|
||
# 解析上下文（复用 CurrentRoleMiddleware 的 request 属性）
|
||
tenant_uuid = getattr(request, 'current_tenant_uuid', None)
|
||
company_uuid = getattr(request, 'current_company_uuid', None)
|
||
user = getattr(request, 'user', None)
|
||
user_uuid = getattr(user, 'UserUUID', None) if user and hasattr(user, 'UserUUID') else None
|
||
|
||
# 请求头摘要（仅保留安全相关头，避免记录敏感信息）
|
||
safe_headers = {}
|
||
for key in ('HTTP_USER_AGENT', 'HTTP_REFERER', 'HTTP_ORIGIN', 'HTTP_HOST', 'CONTENT_TYPE'):
|
||
val = request.META.get(key, '')
|
||
if val:
|
||
safe_headers[key] = val[:256]
|
||
|
||
# 请求体摘要
|
||
body_summary = ''
|
||
if hasattr(request, 'body'):
|
||
try:
|
||
raw = request.body
|
||
if raw:
|
||
body_summary = raw[:self.max_body_len].decode('utf-8', errors='replace')
|
||
except Exception:
|
||
body_summary = '<binary>'
|
||
|
||
attempt_type = self._classify(response.status_code)
|
||
|
||
AbnormalUserLog.objects.using('gerpcore').create(
|
||
LogUUID=uuid.uuid4().bytes,
|
||
TenantUUID_id=tenant_uuid,
|
||
CompanyUUID_id=company_uuid,
|
||
UserUUID_id=user_uuid,
|
||
AccountID=str(user_uuid or '')[:64],
|
||
IPAddress=request.META.get('REMOTE_ADDR', '')[:64],
|
||
RealIP=request.META.get('HTTP_X_REAL_IP', '')[:64],
|
||
XForwardedFor=request.META.get('HTTP_X_FORWARDED_FOR', '')[:512],
|
||
UserAgent=request.META.get('HTTP_USER_AGENT', '')[:512],
|
||
RequestMethod=request.method[:16],
|
||
RequestPath=request.path[:512],
|
||
RequestParams=dict(request.GET) if request.GET else None,
|
||
RequestBody=body_summary,
|
||
RequestHeaders=safe_headers or None,
|
||
AttemptType=attempt_type,
|
||
ResponseStatus=response.status_code,
|
||
Remark=f'auto-detected by AttackDetectMiddleware',
|
||
)
|
||
except Exception:
|
||
logger.warning('[AttackDetect] 记录攻击日志失败', exc_info=True)
|
||
|
||
@staticmethod
|
||
def _classify(status_code):
|
||
"""根据状态码分类攻击类型。"""
|
||
mapping = {
|
||
401: 'auth_failure',
|
||
403: 'permission_denied',
|
||
429: 'rate_limited',
|
||
}
|
||
return mapping.get(status_code, 'suspicious')
|