48 lines
2.9 KiB
Python
48 lines
2.9 KiB
Python
|
import base64
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
import queue
|
||
|
|
import secrets
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import threading
|
||
|
|
import unittest
|
||
|
|
import urllib.error
|
||
|
|
import urllib.request
|
||
|
|
|
||
|
|
|
||
|
|
@unittest.skipUnless(sys.platform=='win32','Production inspector isolation requires Windows')
|
||
|
|
class InspectorHTTPTests(unittest.TestCase):
|
||
|
|
@classmethod
|
||
|
|
def setUpClass(cls):
|
||
|
|
cls.token=secrets.token_urlsafe(32)
|
||
|
|
env={key:value for key,value in os.environ.items() if key.upper() in {'SYSTEMROOT','WINDIR','USERPROFILE','LOCALAPPDATA','APPDATA','TEMP','TMP'}}
|
||
|
|
env.update({'INSPECTOR_TOKEN':cls.token,'INSPECTOR_PORT':'0'})
|
||
|
|
cls.server=subprocess.Popen([sys.executable,str(Path(__file__).with_name('server.py'))],stdout=subprocess.PIPE,stderr=subprocess.PIPE,env=env,creationflags=subprocess.CREATE_NO_WINDOW)
|
||
|
|
lines=queue.Queue();threading.Thread(target=lambda:lines.put(cls.server.stdout.readline()),daemon=True).start()
|
||
|
|
try:
|
||
|
|
cls.ready=json.loads(lines.get(timeout=40));cls.port=cls.ready['port']
|
||
|
|
if cls.ready['containment']['parserIsolation']['kind']!='windows-appcontainer-v1':raise RuntimeError('Production sandbox attestation missing')
|
||
|
|
except Exception:
|
||
|
|
cls.server.kill();cls.server.wait(timeout=5);cls.server.stdout.close();cls.server.stderr.close();raise
|
||
|
|
@classmethod
|
||
|
|
def tearDownClass(cls):
|
||
|
|
cls.server.kill();cls.server.wait(timeout=5);cls.server.stdout.close();cls.server.stderr.close()
|
||
|
|
def request(self,data,credential):
|
||
|
|
policy={'dlp':{'maxBytes':1048576,'rules':[{'id':'secret','kind':'literal','value':'Secret Roadmap'}],'semantic':{'enabled':False}}}
|
||
|
|
body=json.dumps({'data':base64.b64encode(data).decode(),'policy':policy}).encode()
|
||
|
|
request=urllib.request.Request(f'http://127.0.0.1:{self.port}/inspect',data=body,headers={'Content-Type':'application/json','Authorization':'Bearer '+credential})
|
||
|
|
return urllib.request.urlopen(request,timeout=30)
|
||
|
|
def test_authenticated_content_inspection(self):
|
||
|
|
with self.request(b'Secret Roadmap',self.token) as response:
|
||
|
|
result=json.load(response);self.assertEqual(result['decision'],'block');self.assertEqual(result['ruleId'],'secret')
|
||
|
|
with self.request(b'Public information',self.token) as response:self.assertEqual(json.load(response)['decision'],'allow')
|
||
|
|
def test_rejects_unauthenticated_content(self):
|
||
|
|
with self.assertRaises(urllib.error.HTTPError) as error:self.request(b'Public information','wrong')
|
||
|
|
self.assertEqual(error.exception.code,401);error.exception.close()
|
||
|
|
def test_actual_parser_isolation_attestation(self):
|
||
|
|
isolation=self.ready['containment']['parserIsolation'];self.assertTrue(isolation['active']);self.assertEqual(isolation['networkCapabilities'],0);self.assertTrue(isolation['privatePipes']);self.assertTrue(isolation['readOnlyRuntime'])
|
||
|
|
|
||
|
|
|
||
|
|
if __name__=='__main__':unittest.main()
|