first commit

apps/inspector/test_server.py

@@ -0,0 +1,48 @@
+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()