Archived
StasonJatham
5c46b47937
Preserve the GitHub history and the newer deployed source from ct_134_karldns. Exclude runtime data, credentials, AppleDouble files and local backup copies. Make Gitea the canonical source for documentation, package metadata and project links.
164 lines
11 KiB
Python
164 lines
11 KiB
Python
"""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))
|