39 lines
2.6 KiB
Python
39 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
import hashlib,ipaddress,json,os,socket,struct,time
|
|
from pathlib import Path
|
|
SIZE=109688;SHA="25b972ac202050ab5a9c1c89651b27e841cdf788d19c93bf75f051c9c8acd403";REC=struct.Struct("<8sIIiiIIQQQqq");PORT=9021
|
|
FIELDS={"active","run_id","target","port","artifact_size","artifact_sha256","output_path","receipt_path","not_before","not_after","one_connection","one_transfer","one_execution","result_receive","two_exact_lstat","file_content_read","device_write","retry","reconnect"}
|
|
class Error(RuntimeError):pass
|
|
def load(p):return json.loads(p.read_text())
|
|
def validate(a,b,now):
|
|
if set(a)!=FIELDS or set(b)!=FIELDS or a!=b or a["active"] is not True:raise Error("inactive/mismatch")
|
|
if not a["not_before"]<=now<=a["not_after"] or a["port"]!=PORT or a["artifact_size"]!=SIZE or a["artifact_sha256"]!=SHA:raise Error("window/identity")
|
|
if any(a[k] is not True for k in ("one_connection","one_transfer","one_execution","result_receive","two_exact_lstat")) or any(a[k] is not False for k in ("file_content_read","device_write","retry","reconnect")):raise Error("authority")
|
|
ipaddress.IPv4Address(a["target"])
|
|
for k in ("output_path","receipt_path"):
|
|
p=Path(a[k]);
|
|
if not p.is_absolute() or p.exists():raise Error("exclusive path")
|
|
return a
|
|
def put(p,v):
|
|
with p.open("x",encoding="utf-8",newline="\n") as f:json.dump(v,f,sort_keys=True);f.write("\n");f.flush();os.fsync(f.fileno())
|
|
def run(elf,activation,approval,factory=socket.socket,wall=time.time):
|
|
p=validate(load(activation),load(approval),wall());raw=elf.read_bytes()
|
|
if len(raw)!=SIZE or hashlib.sha256(raw).hexdigest()!=SHA:raise Error("artifact")
|
|
put(Path(p["receipt_path"]),{"consumed_before_socket":True,"run_id":p["run_id"],"artifact_sha256":SHA});s=None;wire=bytearray()
|
|
try:
|
|
s=factory(socket.AF_INET,socket.SOCK_STREAM);s.settimeout(3);s.connect((p["target"],PORT));s.sendall(raw);s.shutdown(socket.SHUT_WR);s.settimeout(10)
|
|
while len(wire)<144:
|
|
c=s.recv(144-len(wire))
|
|
if not c:break
|
|
wire.extend(c)
|
|
if len(wire)!=144 or s.recv(1):raise Error("result length")
|
|
result=[]
|
|
for i in range(2):
|
|
v=REC.unpack(wire[i*72:(i+1)*72])
|
|
if v[0]!=b"CHP10DU1" or v[1]!=1 or v[2]!=i or v[6]!=0:raise Error("record")
|
|
result.append({"index":i,"exists":bool(v[3]),"errno":v[4],"mode":v[5],"size":v[7],"device":v[8],"inode":v[9],"mtime":v[10],"ctime":v[11]})
|
|
put(Path(p["output_path"]),{"paths":["/user/app/FAKE00000/app.pkg","/mnt/usb0/IV9999-FAKE00000_00-HOMEBREWLOADER01.pkg"],"records":result,"run_id":p["run_id"]});return result
|
|
finally:
|
|
if s is not None:s.close()
|