"""Nuno development API client. Python 3.11+ and cryptography==46.0.3. No shell/server/VPN account. Keys are generated and retained only on this device. """ import argparse,base64,getpass,hashlib,json,os,secrets,subprocess,sys,time,urllib.request,urllib.parse from pathlib import Path from cryptography.hazmat.primitives import hashes,serialization from cryptography.hazmat.primitives.asymmetric import ec ORIGIN='https://medical.polymorphic.group' HOME=Path.home()/'.medical-development' class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self,*a,**kw): return None HTTP=urllib.request.build_opener(urllib.request.ProxyHandler({}),NoRedirect()) def protect(): import stat HOME.mkdir(mode=0o700,exist_ok=True) paths=[HOME]+[HOME/x for x in ('device-key.pem','device.json') if (HOME/x).exists() or (HOME/x).is_symlink()] for path in paths: st=path.lstat() if stat.S_ISLNK(st.st_mode) or getattr(st,'st_file_attributes',0)&0x400: raise RuntimeError('Reparse points and symlinks forbidden') if path==HOME and not stat.S_ISDIR(st.st_mode): raise RuntimeError('Configuration must be a directory') if path!=HOME and not stat.S_ISREG(st.st_mode): raise RuntimeError('Configuration must be regular files') if os.name=='nt': # Replace the DACL instead of retaining potentially permissive explicit ACEs. for path in paths: literal=str(path).replace("'","''") isdir=path==HOME kind='DirectorySecurity' if isdir else 'FileSecurity' inherit='ContainerInherit,ObjectInherit' if isdir else 'None' script="$sid=[System.Security.Principal.WindowsIdentity]::GetCurrent().User; $acl=New-Object System.Security.AccessControl."+kind+"; $acl.SetAccessRuleProtection($true,$false); $rule=New-Object System.Security.AccessControl.FileSystemAccessRule($sid,'FullControl','"+inherit+"','None','Allow'); $acl.AddAccessRule($rule); Set-Acl -LiteralPath '"+literal+"' -AclObject $acl -ErrorAction Stop" subprocess.run(['powershell.exe','-NoProfile','-NonInteractive','-Command',script],check=True,stdout=subprocess.DEVNULL) else: for path in paths: if path.stat().st_uid!=os.getuid(): raise RuntimeError('Wrong configuration owner') os.chmod(path,0o700 if path==HOME else 0o600) def write_new(path,raw): fd=os.open(path,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600) with os.fdopen(fd,'wb') as f: f.write(raw); f.flush(); os.fsync(f.fileno()) def key(): path=HOME/'device-key.pem' if not path.exists(): k=ec.generate_private_key(ec.SECP256R1()) write_new(path,k.private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.PKCS8,serialization.NoEncryption())) if path.is_symlink(): raise RuntimeError('Key must not be a symlink') return serialization.load_pem_private_key(path.read_bytes(),password=None) def sign(k,raw): return base64.b64encode(k.sign(raw,ec.ECDSA(hashes.SHA256()))).decode() def request(path,body=None,headers=None): raw=None if body is None else json.dumps(body,separators=(',',':')).encode() h={'Content-Type':'application/json'}; h.update(headers or {}) with HTTP.open(urllib.request.Request(ORIGIN+path,data=raw,headers=h),timeout=105) as r: result=r.read(4*1024*1024+1) if len(result)>4*1024*1024: raise RuntimeError('Response too large') return json.loads(result) def call(path,body=None): # No automatic retry: a lost response may already have invoked inference. k=key(); c=json.loads((HOME/'device.json').read_text()); stamp=str(int(time.time())); nonce=secrets.token_hex(24) raw=b'' if body is None else json.dumps(body,separators=(',',':')).encode(); method='GET' if body is None else 'POST' signed=('medical-device-v1\n'+ORIGIN+'\n'+method+'\n'+path+'\n'+stamp+'\n'+nonce+'\n'+hashlib.sha256(raw).hexdigest()).encode() return request(path,body,{'X-Device':c['device'],'X-Time':stamp,'X-Nonce':nonce,'X-Signature':sign(k,signed)}) def enroll(): if (HOME/'device.json').exists(): raise RuntimeError('Device already enrolled; use models or test') link=getpass.getpass('Paste the complete private onboarding link (hidden): ').strip() u=urllib.parse.urlsplit(link) if u.scheme!='https' or u.netloc!='medical.polymorphic.group' or u.path!='/setup' or u.query or not u.fragment: raise RuntimeError('Unexpected onboarding link') token=u.fragment password=getpass.getpass('Invitation password (international cellphone number): ').strip() k=key(); public=k.public_key().public_bytes(serialization.Encoding.PEM,serialization.PublicFormat.SubjectPublicKeyInfo).decode() message=('medical-enroll-v1\n'+ORIGIN+'\n'+token+'\n'+public).encode() result=request('/enroll',{'invite':token,'password':password,'public':public,'signature':sign(k,message)}) write_new(HOME/'device.json',json.dumps(result).encode()) print('Device enrolled. Access expires UTC:',time.strftime('%Y-%m-%d %H:%M:%S',time.gmtime(result['expires']))) print('Run models, then test. Credentials remain in',HOME) def main(): p=argparse.ArgumentParser(description=__doc__); p.add_argument('action',choices=['enroll','models','test','chat','serve']); p.add_argument('--input',help='JSON request file for chat'); p.add_argument('--port',type=int,default=18195); a=p.parse_args(); protect() if a.action=='enroll': return enroll() if a.action=='models': result=call('/v1/models') elif a.action=='test': result=call('/v1/chat/completions',{'model':'qwen3.8-27b','messages':[{'role':'user','content':'Synthetic test: what is 17 multiplied by 23? Reply with the number only.'}],'max_tokens':16,'temperature':0,'stream':False}) elif a.action=='chat': if not a.input: p.error('chat requires --input JSONFILE') raw=Path(a.input).read_bytes() if len(raw)>65536: raise RuntimeError('Request too large') result=call('/v1/chat/completions',json.loads(raw)) else: from http.server import BaseHTTPRequestHandler,HTTPServer # A random local bearer credential prevents unrelated web pages from invoking it. local=secrets.token_urlsafe(32) class Local(BaseHTTPRequestHandler): def log_message(self,*a): pass def setup(self): super().setup(); self.connection.settimeout(10) def send(self,code,obj): raw=json.dumps(obj).encode(); self.send_response(code); self.send_header('Content-Type','application/json'); self.send_header('Content-Length',str(len(raw))); self.send_header('Cache-Control','no-store'); self.send_header('Connection','close'); self.end_headers(); self.wfile.write(raw); self.close_connection=True def run_request(self): if self.headers.get('Host')!='127.0.0.1:'+str(a.port) or self.headers.get('Origin') or self.headers.get('Sec-Fetch-Site') or self.headers.get('Authorization')!='Bearer '+local: return self.send(403,{'error':'local_auth_required'}) if self.path not in ('/v1/models','/v1/chat/completions'): return self.send(404,{'error':'not_found'}) body=None if self.command=='POST': v=self.headers.get_all('Content-Length',[]) if len(v)!=1 or not v[0].isdigit() or not 1<=int(v[0])<=65536 or self.headers.get('Transfer-Encoding') or self.headers.get('Content-Encoding') or self.headers.get_content_type()!='application/json': return self.send(400,{'error':'invalid_request'}) raw=self.rfile.read(int(v[0])) try: body=json.loads(raw) except Exception: return self.send(400,{'error':'invalid_json'}) if (self.command,self.path) not in [('GET','/v1/models'),('POST','/v1/chat/completions')]: return self.send(405,{'error':'method'}) try: self.send(200,call(self.path,body)) except urllib.error.HTTPError as e: self.send(e.code,{'error':'remote_request_rejected'}) except Exception: self.send(503,{'error':'connection_failed_no_automatic_retry'}) do_GET=run_request; do_POST=run_request print('Application base URL: http://127.0.0.1:'+str(a.port)+'/v1',flush=True) print('Local application API key (this process only): '+local,flush=True) print('Keep this terminal private. Ctrl+C stops the connector. No inbound firewall change.',flush=True) HTTPServer(('127.0.0.1',a.port),Local).serve_forever(); return print(json.dumps(result,indent=2)) if __name__=='__main__': try: main() except KeyboardInterrupt: pass except Exception as e: # Never print request objects, private link, keys or supplied password. print('Failed: '+type(e).__name__+'. No automatic retries. Check instructions; keep any existing device-key.pem for enrolment recovery.',file=sys.stderr); sys.exit(1)