73 lines
4.7 KiB
Python
73 lines
4.7 KiB
Python
"""Run: python3 test_nitro.py. Inert API behaviour and exact upload capture."""
|
|
import base64
|
|
import hashlib
|
|
import http.client
|
|
import json
|
|
import pathlib
|
|
import tempfile
|
|
import threading
|
|
import urllib.parse
|
|
from types import SimpleNamespace
|
|
from evidence import Evidence
|
|
from honeypot import Engine, Server
|
|
from nitro import CONFIG, RESOURCES, respond
|
|
|
|
|
|
def main():
|
|
state = SimpleNamespace()
|
|
auth = {}
|
|
def api(method, path, data=None, headers=None, source=state):
|
|
return respond(source, method, urllib.parse.urlsplit('/nitro/v1/'+path),
|
|
json.dumps(data).encode() if data is not None else b'', headers if headers is not None else auth)
|
|
assert api('GET', 'config/nsversion')[0] == 401
|
|
assert api('POST', 'config/login', {'login': {}})[0] == 400
|
|
status, login, extra = api('POST', 'config/login', {'login': {'username':'probe','password':'inert'}})
|
|
assert status == 201 and extra[0][0] == 'Set-Cookie'
|
|
auth = {'Cookie': 'NITRO_AUTH_TOKEN='+login['sessionid']}
|
|
assert api('GET', 'config/nsversion')[1]['nsversion'][0]['version'].startswith('NetScaler NS')
|
|
assert len(RESOURCES['config']) > 1400 and len(RESOURCES['stat']) > 150
|
|
for kind in RESOURCES:
|
|
for resource in RESOURCES[kind]:assert api('GET',kind+'/'+resource)[0] == 200, resource
|
|
for resource in CONFIG:assert resource in api('GET', 'config/'+resource)[1]
|
|
assert api('POST', 'config/lbvserver', {'lbvserver': {'name':'new','port':80}})[0] == 201
|
|
assert api('GET', 'config/lbvserver/new')[1]['lbvserver'][0]['port'] == 80
|
|
assert api('PUT', 'config/lbvserver', {'lbvserver': {'name':'new','port':8080}})[0] == 200
|
|
assert api('GET', 'config/lbvserver/new')[1]['lbvserver'][0]['port'] == 8080
|
|
assert api('POST', 'config/lbvserver?action=disable', {'lbvserver': {'name':'new'}})[0] == 200
|
|
assert api('GET', 'config/lbvserver/new')[1]['lbvserver'][0]['state'] == 'DISABLED'
|
|
assert api('GET', 'config/lbvserver?filter=name:new&attrs=name')[1]['lbvserver'] == [{'name':'new'}]
|
|
assert api('GET', 'config/lbvserver?count=yes')[1]['lbvserver'] == [{'__count':2}]
|
|
assert api('GET', 'config/lbvserver_binding/lb_apps')[1]['lbvserver_binding'][0]['lbvserver_service_binding'][0]['servicename'] == 'svc_app_https'
|
|
assert api('GET', 'config/lbvserver?pagesize=1&pageno=2')[1]['lbvserver'][0]['name'] == 'new'
|
|
assert api('GET', 'config/lbvserver?pagesize=bad')[0] == 400
|
|
assert api('GET', 'config/nsip?'+ '&'.join('x=1' for _ in range(129)))[0] == 400
|
|
assert api('GET', 'config/not_a_resource')[0] == 404
|
|
assert api('DELETE', 'config/lbvserver/new')[0] == 200
|
|
assert api('GET', 'config/lbvserver/new')[1]['lbvserver'] == []
|
|
assert api('GET', 'config/lbvserver', source=SimpleNamespace(), headers={'X-NITRO-USER':'probe','X-NITRO-PASS':'inert'})[1]['lbvserver'] == CONFIG['lbvserver']
|
|
assert api('POST', 'config/lbvserver', {'lbvserver':{'name':'huge','blob':'x'*65520}})[0] in (400,503)
|
|
assert api('POST', 'config/logout', {'logout':{}})[0] == 201
|
|
assert api('GET', 'config/nsversion')[0] == 401
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
evidence = Evidence(directory)
|
|
server = Server(('127.0.0.1',0), Engine(evidence.emit,save=evidence.save), serve_guest_files=False)
|
|
thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
|
|
payload=b'\0inert-nitro-file\xff\n'
|
|
body=json.dumps({'systemfile':{'filename':'../../inert.bin','filelocation':'/tmp','filecontent':base64.b64encode(payload).decode()}}).encode()
|
|
try:
|
|
connection=http.client.HTTPConnection('127.0.0.1',server.server_port,timeout=3)
|
|
connection.request('POST','/nitro/v1/config/systemfile',body,{'Content-Type':'application/json','X-NITRO-USER':'probe','X-NITRO-PASS':'inert'})
|
|
response=connection.getresponse();assert response.status==201;response.read();connection.close()
|
|
digest=hashlib.sha256(payload).hexdigest()
|
|
assert (pathlib.Path(directory)/'artifacts'/(digest+'.bin')).read_bytes()==payload
|
|
records=[json.loads(row) for row in (pathlib.Path(directory)/'events.jsonl').read_text().splitlines()]
|
|
upload=next(row for row in records if row['event']=='upload')
|
|
assert upload['filename']=='../../inert.bin' and upload['evidence']['sha256']==digest
|
|
assert any(row['event']=='request_body' and row['evidence']['sha256']==hashlib.sha256(body).hexdigest() for row in records)
|
|
finally:
|
|
server.shutdown();server.server_close();thread.join(3);evidence.close()
|
|
print('PASS: resource catalog, auth, CRUD, actions, queries, source isolation and byte-exact NITRO files')
|
|
|
|
|
|
if __name__=='__main__':main()
|