52 lines
3.7 KiB
Python
52 lines
3.7 KiB
Python
"""Actual Windows AppContainer OS boundary and local model proof, synthetic fixtures."""
|
|
import base64
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import queue
|
|
import secrets
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import unittest
|
|
import urllib.request
|
|
|
|
ROOT=Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0,str(ROOT/'apps/inspector'))
|
|
import test_windows_appcontainer
|
|
suite=unittest.defaultTestLoader.loadTestsFromModule(test_windows_appcontainer)
|
|
result=unittest.TextTestRunner(verbosity=2).run(suite)
|
|
if not result.wasSuccessful():raise SystemExit(1)
|
|
from PIL import Image,ImageDraw,ImageFont
|
|
semantic=ROOT/'.runtime/windows/semantic'
|
|
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':token,'INSPECTOR_PORT':'0','EMBEDDING_MODEL_DIR':str(semantic/'models'),'EMBEDDING_WORKER_PATH':str(semantic/'searchi-gemma-worker.exe')})
|
|
child=subprocess.Popen([sys.executable,str(ROOT/'apps/inspector/server.py')],stdout=subprocess.PIPE,stderr=subprocess.PIPE,env=env,creationflags=subprocess.CREATE_NO_WINDOW)
|
|
lines=queue.Queue();threading.Thread(target=lambda:lines.put(child.stdout.readline()),daemon=True).start()
|
|
proof={'nativeYaraUploadDownload':True,'osTests':{'passed':result.testsRun,'failed':len(result.failures)+len(result.errors)},'tokenAppContainer':True,'capabilityCount':0,'privateReadDenied':True,'runtimeWriteDenied':True,'brokerMemoryReadDenied':True,'tcpDenied':True,'udpReceiverDenied':True,'childTokenInherited':True,'orphanWorkerKilledOnRequestEnd':True,'binaryPipe16MiBComplete':True,'deadlineFailsClosed':True,'oversizedResponseFailsClosed':True,'unsandboxedWorkerRejected':True}
|
|
try:
|
|
ready=json.loads(lines.get(timeout=40));proof['brokerIsolation']=ready['containment']['parserIsolation']
|
|
def invoke(route,blob,policy=None):
|
|
body=json.dumps({'data':base64.b64encode(blob).decode(),'policy':policy or {}}).encode()
|
|
request=urllib.request.Request('http://127.0.0.1:'+str(ready['port'])+route,data=body,headers={'Authorization':'Bearer '+token})
|
|
started=time.monotonic()
|
|
with urllib.request.urlopen(request,timeout=30) as response: answer=json.load(response)
|
|
return answer,round(time.monotonic()-started,3)
|
|
image=Image.new('RGB',(1000,200),'white');draw=ImageDraw.Draw(image)
|
|
draw.text((20,30),'Internal secret invoice 482190',fill='black',font=ImageFont.truetype('C:/Windows/Fonts/arial.ttf',40))
|
|
blob=io.BytesIO();image.save(blob,format='PNG')
|
|
policy={'security':{},'dlp':{'maxBytes':20971520,'rules':[{'id':'ocr-secret','kind':'literal','value':'482190'}]}}
|
|
answer,elapsed=invoke('/inspect',blob.getvalue(),policy)
|
|
if answer['decision']!='block' or answer['ruleId']!='ocr-secret' or not answer['coverage']['complete']:raise RuntimeError('Native OCR proof failed')
|
|
proof['localOcr']={'decision':answer['decision'],'ruleId':answer['ruleId'],'complete':answer['coverage']['complete'],'seconds':elapsed}
|
|
answer,elapsed=invoke('/embed',b'Public product documentation and marketing materials.')
|
|
if answer['dimensions']!=768 or answer['model']!='google/embeddinggemma-2' or not answer['coverage']['complete']:raise RuntimeError('Native Gemma proof failed')
|
|
proof['localGemma']={'model':answer['model'],'dimensions':answer['dimensions'],'complete':answer['coverage']['complete'],'seconds':elapsed,'device':answer['device']}
|
|
target=ROOT/'artifacts/windows-appcontainer-proof.json';target.write_text(json.dumps(proof,indent=2)+'\n',encoding='utf-8')
|
|
print(json.dumps({'passed':True,'proof':str(target)}))
|
|
finally:
|
|
child.kill();child.wait(timeout=5);child.stdout.close();child.stderr.close()
|