karldns.de/karldns/web/request_capture.py

"""Bounded inert request evidence for explicitly authorized cloud analysis; never execute it."""
import ipaddress,json,re
from urllib.parse import unquote,unquote_plus,parse_qsl,urlencode

VERSION=1
SENSITIVE=re.compile(r'password|passwd|pwd|secret|token|authorization|api.?key|auth.?key|session|cookie|credential|private.?key|client.?secret|csrf|xsrf|nonce|^(?:user(?:name)?|email|login|sid)$',re.I)

def clean(text,secrets=()):
 text=str(text)
 for secret in secrets:
  if isinstance(secret,str) and len(secret)>=1:text=text.replace(secret,'[SECRET]')
 text=re.sub(r'-----BEGIN [^-]*PRIVATE KEY-----.*?-----END [^-]*PRIVATE KEY-----','[PRIVATE_KEY]',text,flags=re.S)
 text=re.sub(r'\b(?:Bearer|Basic)\s+[^\s"\'<>]+','[AUTH]',text,flags=re.I)
 # Quoted JSON, forms, XML attributes and common key=value credentials, including malformed payloads.
 text=re.sub(r'''(?ix)(["']?(?:password|passwd|pwd|credential|private[_-]?key|secret|[\w-]*token|authorization|(?:api|access|auth)[_-]?key|client[_-]?secret|session(?:[_-]?id)?|cookie|user(?:name)?|login|email|csrf|xsrf|nonce|sid)["']?\s*[:=]\s*)("(?:\\.|[^"\\])*"|'(?:\\.|[^'\\])*'|[^\s&;,<>}\]]+)''',lambda m:m[1]+'"[SECRET]"',text)
 text=re.sub(r'(?is)(<(?:password|secret|token|user(?:name)?|session)[^>]*>).*?(</[^>]+>)',r'\1[SECRET]\2',text)
 text=re.sub(r'\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b','[JWT]',text)
 text=re.sub(r'\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b','[EMAIL]',text,flags=re.I)
 text=re.sub(r'(?i)(https?://)[^\s/"\'<>]+',r'\1[HOST]',text)
 text=re.sub(r'(?i)\b(?:[a-z0-9-]+\.)+(?:com|net|org|de|fail|local|internal|io|xyz|ru|cn|top|test|invalid)\b','[HOST]',text)
 text=re.sub(r'(?<![\w.])(?:\d{1,3}\.){3}\d{1,3}(?![\w.])','[IP]',text)
 def ipv6(m):
  try:ipaddress.IPv6Address(m[0]);return '[IP]'
  except ValueError:return m[0]
 text=re.sub(r'(?<!\w)(?:[0-9a-fA-F]{0,4}:){2,}[0-9a-fA-F:.]*(?!\w)',ipv6,text)
 text=re.sub(r'(?<![\w])[A-Za-z0-9_+/=-]{32,}(?![\w])',lambda m:'[OPAQUE:'+str(len(m[0]))+']',text)
 text=re.sub(r'(?i)(/(?:m|api/v1/(?:hostnames|router/(?:register|update))))/[^\s?\"\'<>]+',r'\1/[SECRET]',text)
 return ''.join(c if c in '\n\r\t' or ord(c)>=32 else ' ' for c in text)

def private_clean(text,secrets=()):
 original=str(text);decoded=original
 for _ in range(4):
  next_value=unquote(decoded)
  if next_value==decoded:break
  decoded=next_value
 if unquote(decoded)!=decoded:return '[ENCODING_LIMIT]'
 cleaned_original=clean(original,secrets);checked=cleaned_original
 for _ in range(4):checked=unquote(checked)
 return cleaned_original if clean(checked,secrets)==checked else clean(decoded,secrets)

def decoded_key(value):
 for _ in range(4):value=unquote(value)
 return value

def structured(value,secrets,depth=0):
 if depth>12:return '[DEPTH_LIMIT]'
 if isinstance(value,dict):return {private_clean(k,secrets)[:128]:('[SECRET]' if (SENSITIVE.search(decoded_key(k)) or unquote(decoded_key(k))!=decoded_key(k)) else structured(v,secrets,depth+1)) for k,v in list(value.items())[:80]}
 if isinstance(value,list):return [structured(v,secrets,depth+1) for v in value[:80]]
 return private_clean(value,secrets) if isinstance(value,str) else value

def structure_truncated(value,depth=0):
 if depth>12:return True
 if isinstance(value,(dict,list)):
  values=list(value.values()) if isinstance(value,dict) else value
  return len(values)>80 or any(structure_truncated(v,depth+1) for v in values[:80])
 return False

def request_sample(s,body,secrets=()):
 r=s['home']['honeypot']['request'];out=[];budget=4000;header_rows=r.get('headers',[])
 for h in header_rows[:48]:
  name=private_clean(h['name'],secrets)[:128];name=name if re.fullmatch(r"[!#$%&'*+.^_`|~0-9A-Za-z-]+",name) else '[INVALID_HEADER]';value=h['value'];sensitive=bool(SENSITIVE.search(decoded_key(h['name']))) or unquote(decoded_key(h['name']))!=decoded_key(h['name']) or name.lower() in {'x-request-id','x-real-ip','x-forwarded-for','cf-connecting-ip','true-client-ip','forwarded'}
  if name.lower()=='host' and re.fullmatch(r'[A-Za-z0-9._:\[\]-]+',value):sensitive=True
  # NSC_USER can itself carry an exploit; ordinary principal names are masked.
  if name.lower()=='nsc_user' and re.fullmatch(r'[\w@. -]{1,128}',value):sensitive=True
  cleaned='[REDACTED]' if sensitive else private_clean(value,secrets)
  limit=min(1000,budget);out.append({'name':name,'value':cleaned[:limit] if limit else '','original_length':len(value),'truncated':len(cleaned)>limit});budget-=min(len(cleaned),limit)
 payload={'status':'missing_or_unverified','text':''}
 if body is not None:
  capped=False
  try:
   parsed=json.loads(body);capped=structure_truncated(parsed);text=json.dumps(structured(parsed,secrets),ensure_ascii=False)
  except (ValueError,TypeError,RecursionError):
   content_type=' '.join(h['value'] for h in header_rows if h['name'].lower()=='content-type').lower()
   if 'application/x-www-form-urlencoded' in content_type:
    try:pairs=parse_qsl(body,keep_blank_values=True,max_num_fields=100)
    except ValueError:pairs=[]
    text=urlencode([(private_clean(k,secrets),'[SECRET]' if (SENSITIVE.search(decoded_key(k)) or unquote(decoded_key(k))!=decoded_key(k)) else private_clean(v,secrets)) for k,v in pairs]) if pairs else private_clean(body,secrets)
   else:text=private_clean(body,secrets)
  payload={'status':'verified_text','text':text[:4000],'truncated':len(text)>4000 or capped,'original_bytes':len(body.encode())}
 method=s.get('http',{}).get('request',{}).get('method','UNKNOWN')
 if not re.fullmatch('[A-Z]{1,20}',str(method)):method='UNKNOWN'
 uri=private_clean(r.get('uri',''),secrets)
 return {'method':method,'uri':uri[:1600],'uri_truncated':len(uri)>1600,'headers':out,'headers_complete':r.get('headers_complete') is True,'headers_truncated':len(header_rows)>48,'payload':payload}


class PostCaptureMiddleware:
 """Bounded predispatch evidence; replay unchanged bytes to the application."""
 def __init__(self,app,settings):
  self.app=app;self.settings=settings

 async def __call__(self,scope,receive,send):
  if scope['type']!='http' or scope['method']!='POST':
   await self.app(scope,receive,send);return
  import datetime,logging,os,uuid
  from starlette.requests import Request
  from starlette.responses import PlainTextResponse
  from ..security import client_ip
  request=Request(scope);headers=[{'name':k.decode('latin1'),'value':v.decode('latin1')} for k,v in scope['headers']]
  secrets=[v for k,v in self.settings.model_dump().items() if SENSITIVE.search(decoded_key(k)) and isinstance(v,str) and v]
  secrets += [v for k,v in os.environ.items() if SENSITIVE.search(decoded_key(k)) and v]
  secrets.append(self.settings.admin_path)
  supplied=request.headers.get('x-request-id','');rid=supplied if re.fullmatch('[a-fA-F0-9]{32}',supplied) else uuid.uuid4().hex
  scope.setdefault('state',{})['audit_request_id']=rid
  scope['headers']=[(k,v) for k,v in scope['headers'] if k.lower()!=b'x-request-id']+[(b'x-request-id',rid.encode())]
  status=500;body=bytearray();complete=False;oversized=False;disconnected=False
  try:declared=int(request.headers.get('content-length','0'))
  except ValueError:declared=0
  try:
   if declared>self.settings.max_request_body_bytes:oversized=True
   else:
    while True:
     message=await receive()
     if message['type']=='http.disconnect':disconnected=True;break
     chunk=message.get('body',b'');remaining=self.settings.max_request_body_bytes+1-len(body)
     body.extend(chunk[:remaining])
     if len(chunk)>remaining or len(body)>self.settings.max_request_body_bytes:oversized=True;break
     if not message.get('more_body',False):complete=True;break
   async def wrapped_send(message):
    nonlocal status
    if message['type']=='http.response.start':
     status=message['status'];message=dict(message)
     message['headers']=[(k,v) for k,v in message.get('headers',[]) if k.lower()!=b'x-request-id']+[(b'x-request-id',rid.encode())]
    await send(message)
   if oversized:
    await PlainTextResponse('request body too large',status_code=413)(scope,receive,wrapped_send)
   elif disconnected:status=499
   else:
    replayed=False
    async def replay():
     nonlocal replayed
     if not replayed:replayed=True;return {'type':'http.request','body':bytes(body),'more_body':False}
     return await receive()
    await self.app(scope,replay,wrapped_send)
  finally:
   content_type=request.headers.get('content-type','').lower();encoding=request.headers.get('content-encoding','identity').lower()
   reason='verified_text';text=None
   if oversized:reason='oversized_omitted'
   elif not complete:reason='incomplete_omitted'
   elif encoding not in ('','identity'):reason='encoded_omitted'
   elif 'multipart/' in content_type:reason='multipart_omitted'
   elif content_type and not any(x in content_type for x in ('text/','json','xml','x-www-form-urlencoded','javascript')):reason='binary_type_omitted'
   else:
    try:
     text=bytes(body).decode('utf-8')
     if any(ord(c)<32 and c not in '\r\n\t' for c in text):reason='binary_omitted';text=None
    except UnicodeDecodeError:reason='invalid_utf8_omitted'
   source={'home':{'honeypot':{'request':{'headers':headers,'headers_complete':True,'uri':scope.get('path','')+('?' + scope.get('query_string',b'').decode('latin1') if scope.get('query_string') else '')}}},'http':{'request':{'method':'POST'}}}
   sample=request_sample(source,text,secrets)
   sample['schema']=1
   sample['headers_complete']=not sample['headers_truncated'] and not any(h['truncated'] for h in sample['headers'])
   sample['payload'].update(status=reason,original_bytes=len(body) if complete else None,captured_bytes=len(body),declared_bytes=declared if 'content-length' in request.headers and declared>=0 else None,complete=complete,capture_truncated=oversized,encoding=encoding if encoding in ('identity','gzip','deflate','br','') else 'other',truncated=sample['payload'].get('truncated',False))
   row={'@timestamp':datetime.datetime.now(datetime.timezone.utc).isoformat(),'event':{'dataset':'karldns.security','kind':'event','category':['web'],'type':['info'],'action':'karldns.post_request','outcome':'unknown','provider':'karldns-native'},'http':{'request':{'method':'POST','id':rid},'response':{'status_code':status}},'karldns':{'request_sample':sample},'message':'KarlDNS sanitized POST request'}
   decoded=unquote_plus(bytes(body).decode('utf-8',errors='replace')).lower()
   row['labels']={'auth_scope':'public_post','actor_identity':'not_available','indicator_verdict':'review_only','unexpected_body':scope.get('path')=='/' and bool(body),'php_code_marker':'<?php' in decoded or '<?=' in decoded,'shell_substitution_marker':'$(' in decoded or '`' in decoded,'sql_union_marker':bool(re.search(r'\bunion\s+(?:all\s+)?select\b',decoded))}
   try:row['source']={'ip':str(ipaddress.ip_address(client_ip(request,self.settings)))}
   except ValueError:pass
   while len(json.dumps(row,ensure_ascii=True))>18000:
    if sample['payload']['text']:
     sample['payload']['text']=sample['payload']['text'][:len(sample['payload']['text'])//2];sample['payload']['truncated']=True
    elif sample['headers']:
     sample['headers'].pop();sample['headers_truncated']=True;sample['headers_complete']=False
    else:sample['uri']=sample['uri'][:200];sample['uri_truncated']=True;break
   logging.getLogger('karldns.security').warning('KARLDNSSEC %s',json.dumps(row,separators=(',',':'),ensure_ascii=True))