This repository has been archived on 2026-10-28. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files

48 lines
3.6 KiB
Python
Raw Permalink Normal View History

import json
import logging
from fastapi.testclient import TestClient
from karldns import DdnsSettings, make_app
from karldns.web.request_capture import private_clean,request_sample
def test_post_capture(tmp_path,caplog):
secret='runtime-canary-6729'
app=make_app(DdnsSettings(database_path=tmp_path/'post.sqlite3',admin_password=secret,hostname_suffix='example.net'))
def event():
events=[json.loads(r.message.removeprefix('KARLDNSSEC ')) for r in caplog.records if r.message.startswith('KARLDNSSEC ')]
return next(e for e in events if e['event']['action']=='karldns.post_request')
@app.post('/capture-exception')
async def failed():raise RuntimeError('synthetic failure')
with TestClient(app,raise_server_exceptions=False) as client,caplog.at_level(logging.WARNING,logger='karldns.security'):
for path,body,ctype,code in [('/',b'', 'text/plain',303),('/api/v1/hostnames',b'', 'text/plain',201),('/missing',b'<?php eval($_POST[x]); ?> UNION SELECT $(id) %252e%252e%252f', 'text/plain',404),('/',b'\xff', 'text/plain',400),('/missing',b'\x00binary','application/octet-stream',404)]:
caplog.clear();response=client.post(path,content=body,headers={'Content-Type':ctype,'X-Canary':secret,'Authorization':'Bearer auth-canary','Cookie':'session=cookie-canary','X-Request-ID':'f'*32},follow_redirects=False)
assert response.status_code==code
e=event();sample=e['karldns']['request_sample']
assert e['http']['request']['id']==response.headers['x-request-id']=='f'*32
for value in [secret,'auth-canary','cookie-canary']:assert value not in caplog.text
assert sample['payload']['original_bytes']==len(body)
if path=='/missing' and ctype=='text/plain':assert sample['payload']['text']==body.decode()
if body==b'\xff':assert sample['payload']['status']=='invalid_utf8_omitted'
caplog.clear();response=client.post('/capture-exception',content='probe=exception');assert response.status_code==500;assert event()['http']['response']['status_code']==500
assert app.state.store.count_accounts()==2
assert app.state.store.list_records()==[]
for headers,body,status in [({'Content-Length':'20000'},b'', 'oversized_omitted'),({'Content-Encoding':'gzip'},b'private', 'encoded_omitted'),({'Content-Type':'multipart/form-data;boundary=z'},b'password=private','multipart_omitted')]:
caplog.clear();r=client.post('/missing',headers=headers,content=body);s=event()['karldns']['request_sample'];assert s['payload']['status']==status;assert not s['payload']['text']
caplog.clear();r=client.post('/admin',content='password=short-canary',headers={'Content-Type':'application/x-www-form-urlencoded'});assert 'short-canary' not in caplog.text
def test_privacy_and_syntax():
for text in ['{"credential":"short-canary"','session=short-canary','private_key=short-canary','%2570assword=short-canary','password=short-canary','https://host.invalid/m/short-canary']:
assert 'short-canary' not in private_clean(text)
attack='<?php system($_GET[x]); ?>; UNION SELECT 1; $(id); %252e%252e%252f'
assert private_clean(attack)==attack
source={'home':{'honeypot':{'request':{'headers':[],'headers_complete':True,'uri':'/'}}},'http':{'request':{'method':'POST'}}}
for body in ['{"%2570assword":"short-canary"}', '{"private_key":"short-canary"}']:
assert 'short-canary' not in json.dumps(request_sample(source,body))
def test_structural_capture_truncation():
source={'home':{'honeypot':{'request':{'headers':[],'headers_complete':True,'uri':'/'}}},'http':{'request':{'method':'POST'}}}
for value in [list(range(81)),{str(i):i for i in range(81)}]:
assert request_sample(source,json.dumps(value))['payload']['truncated']
assert '%252e%252e%252f' in private_clean('password=short-canary&attack=%252e%252e%252f')