`nx_model_get.nx` source

↩ module page · 395 lines · 19427 B
1// nx_model_get.nx -- G9 mg_fetch: hash-verified model acquisition into the gen model store (/compare/gen watch G9).
2// usage: nx_model_get <url> <expected-sha256> <dest-name>
3//   <dest-name> is relative to the model-store root (e.g. diffusion/z_image_turbo-Q8_0.gguf, loras/x.safetensors).
4//
5// COMPOSES INCUMBENTS, DUPLICATES NONE:
6//   transport  = nx_https_fetch_range (own TLS-1.3, Range windows -- one bounded window in memory at a time, so a
7//                multi-GB body STREAMS and can never be silently truncated by a buffer cap)
8//   hashing    = the streaming sha256_init/update/final already inside this compilation unit (nx_sha256 via the
9//                TLS stack / nx_weights_pull) -- the digest covers the FULL body across all windows
10//   verify law = nx_weights_pull (wp_hexenc + fail-closed compare): a mismatched body is DELETED, never stored
11//   decisions  = nx_model_get_lib (dest admission, hex normalize, Content-Range/Length parse, conf parse),
12//                offline-proven by nx_model_get_gate
13// Redirects (HF resolve -> CDN) are followed by a bytes=0-0 probe that reads the Location header, because the
14// range fetcher itself deliberately does not follow. Resumable: an interrupted pull leaves <dest>.part; a re-run
15// re-hashes the part from disk and continues from its byte count. Idempotent: an existing verified dest returns
16// OK-ALREADY-VERIFIED with no network touched. An existing dest that does NOT match is REFUSED and kept
17// (additive-only: this organ never deletes a model already in the store; only its own .part on hash mismatch).
18// exit: 0 VERIFIED / OK-ALREADY-VERIFIED | 1 REFUSED (hash mismatch / existing mismatch) | 2 usage/bad-args |
19//       3 UNPROVEN (transport, trust store, no-length -- partial kept for resume where one exists)
20// license_tier: ORIGINAL
21import "nx_syscalls.nx"
22import "nx_https_fetch_follow.nx"
23import "nx_trust_store_load_from_certdata.nx"
24import "nx_weights_pull.nx"
25import "nx_model_get_lib.nx"
26
27const MG_URLCAP: i64 = 4096        // same bound the fetch stack uses for a url buffer
28const MG_PROBE_CAP: i64 = 262144   // bytes=0-0 probe response buffer (headers + stub body)
29const MG_HOPS_MAX: i64 = 6         // redirect budget, mirrors the fetch stack's default
30const MG_HASH_WIN: i64 = 1048576   // on-disk re-hash window for idempotency/resume
31const MG_WINDOW_DEFAULT: i64 = 8388608   // range window when the conf carries none (8 MiB)
32const MG_WIN_RETRIES: i64 = 3      // window failures tolerated before declaring TRANSPORT-STALL
33const MG_PROGRESS_EVERY: i64 = 32  // announce progress every N windows
34const MG_CERT_ROOTS: i64 = 512
35const MG_CERT_CAP: i64 = 4194304
36const MG_CONF_PATH: *u8 = "knowledge/gen_model_store.conf"
37const MG_ROOT_DEFAULT: *u8 = "/volume1/ai/models"
38const MG_RECEIPT_PATH: *u8 = "knowledge/status/modelget.log"
39
40func mgo_put(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
41func mgo_num(v: i64) -> i64 {
  var m: i64=v
  if m<0 { mgo_put("-" as *u8); m=0-m }
  let t: *u8=sys_mmap(28)
  var k: i64=0
  if m==0 { t[0]=48 as u8; k=1 }
  while m>0 { t[k]=(48+(m%10)) as u8; m=m/10; k=k+1 }
  let b: *u8=sys_mmap(28)
  var j: i64=0
  while j<k { b[j]=t[k-1-j]; j=j+1 }
  sys_write(1,b,k)
  return 0
53}
54
55// stream-hash an on-disk file into ctx (sha256_update over MG_HASH_WIN windows). Returns bytes hashed, -1 unreadable.
56func mg_hash_file_into(ctx: *Sha256, path: *u8) -> i64 {
  let fd: i64 = sys_openat_rd(path)
  if fd<0 { return 0-1 }
  let buf: *u8 = sys_mmap(MG_HASH_WIN)
  var total: i64=0
  var go: i64=1
  var bad: i64=0
  while go==1 {
      let r: i64 = sys_read(fd, buf, MG_HASH_WIN)
      if r<0 { bad=1; go=0 } else { if r==0 { go=0 } else { sha256_update(ctx, buf, r); total=total+r } }
  }
  sys_close(fd)
  if bad==1 { return 0-1 }
  return total
70}
71
72// full-file sha256 hex of an existing artifact (for the idempotency check). Returns bytes, -1 unreadable.
73func mg_hash_file_hex(path: *u8, out_hex: *u8) -> i64 {
  let ctx_raw: *u8 = sys_mmap(256)
  let ctx: *Sha256 = ctx_raw as *Sha256
  sha256_init(ctx)
  let n: i64 = mg_hash_file_into(ctx, path)
  if n<0 { return 0-1 }
  let dig: *u8 = sys_mmap(32)
  sha256_final(ctx, dig)
  wp_hexenc(dig, 32, out_hex)
  return n
83}
84
85// ONE bytes=0-0 range request with HEADER visibility (status, Location, Content-Range total, Content-Length).
86// The shipped nx_https_fetch_range returns only the body, so redirect-following a range pull needs this probe.
87// Returns: 3 REDIRECT (locout filled) | 2 RANGE-OK (totbox = total from Content-Range) |
88//          1 NO-RANGE (200; totbox = Content-Length or -1) | negative = transport/parse failure.
89func mg_probe(urlbuf: *u8, store: *TrustStore, use_chrome: i64, st: *i64, locout: *u8, totbox: *i64) -> i64 {
  st[0]=0
  totbox[0]=0-1
  let target_raw: *u8 = sys_mmap(64)
  let target: *NxHttpsTarget = target_raw as *NxHttpsTarget
  target.url = nx_url_new()
  target.port = 0
  if nx_https_url_for_fetch(urlbuf, target) != NX_HTTPS_URL_OK { return 0-1 }
  let fd_p: *i64 = (sys_mmap(16)) as *i64
  if nx_https_url_connect(target, urlbuf, sys_now_realtime_sec(), fd_p) != NX_HTTPS_CONNECT_OK { return 0-2 }
  let fd: i64 = fd_p[0]
  sys_set_socket_timeout(fd, 30)
  let cr: *u8 = sys_mmap(32)
  let priv: *u8 = sys_mmap(32)
  nx_csprng_fill(cr, 32)
  nx_csprng_fill(priv, 32)
  let vc_raw: *u8 = sys_mmap(64)
  let vc: *TlsValidationContext = vc_raw as *TlsValidationContext
  vc.store = store
  vc.sni_host = urlbuf + target.url.host_off
  vc.sni_host_len = target.url.host_len
  vc.now_epoch = sys_now_realtime_sec()
  var sr: i64 = 0
  if use_chrome==1 { sr = nx_tls13_client_session_run_chrome(fd, urlbuf + target.url.host_off, target.url.host_len, cr, priv, vc) }
  else { sr = nx_tls13_client_session_run(fd, urlbuf + target.url.host_off, target.url.host_len, cr, priv, vc) }
  if sr<=0 { sys_close(fd); return 0-3 }
  let session: *Tls13ClientSession = sr as *Tls13ClientSession
  let path: *u8 = sys_mmap(2048)
  let plen: i64 = ff_path(urlbuf, target, path)
  let req: *u8 = sys_mmap(4096)
  let req_len: i64 = nx_http_build_range_get(path, plen, urlbuf + target.url.host_off, target.url.host_len, 0, 0, req)
  let buf: *u8 = sys_mmap(MG_PROBE_CAP)
  let gc: i64 = nx_https_req_complete(session, fd, req, req_len, buf, MG_PROBE_CAP)
  sys_close(fd)
  if gc<0 { return 0-4 }
  let r: *i64 = (sys_mmap(128)) as *i64
  if nx_http_response_parse(buf, gc, r) != 0 { return 0-5 }
  st[0] = r[1]
  let body_off: i64 = r[6]
  if nx_redir_is_redirect(r[1])==1 {
      let loc: *u8 = sys_mmap(MG_URLCAP)
      let ln: i64 = nx_redir_location(buf, gc, loc, MG_URLCAP)
      if ln<=0 { return 0-6 }
      let resolved: *u8 = sys_mmap(MG_URLCAP)
      ff_resolve_location(loc, urlbuf, target, resolved)
      var k: i64=0
      while resolved[k]!=(0 as u8) { locout[k]=resolved[k]; k=k+1 }
      locout[k]=0 as u8
      return 3
  }
  if r[1]==206 { totbox[0] = mg_content_range_total(buf, body_off); return 2 }
  if r[1]==200 { totbox[0] = mg_content_length(buf, body_off); return 1 }
  return 0-7
142}
143
144// follow redirects (bounded) until a range-answering endpoint; urlbuf ends holding the FINAL url.
145// chromebox carries/returns the TLS-hello flavor that worked. Returns probe mode (1/2) or negative.
146func mg_resolve(urlbuf: *u8, store: *TrustStore, chromebox: *i64, totbox: *i64) -> i64 {
  let st: *i64 = (sys_mmap(16)) as *i64
  let loc: *u8 = sys_mmap(MG_URLCAP)
  var hop: i64=0
  var lastrc: i64=0-9
  while hop<=MG_HOPS_MAX {
      var rc: i64 = mg_probe(urlbuf, store, chromebox[0], st, loc, totbox)
      if rc<0 { if chromebox[0]==0 {
          chromebox[0]=1
          rc = mg_probe(urlbuf, store, 1, st, loc, totbox)
      } }
      if rc==3 {
          mgo_put("hop=" as *u8); mgo_num(hop+1); mgo_put(" status=" as *u8); mgo_num(st[0]); mgo_put(" -> " as *u8); mgo_put(loc); mgo_put("\n" as *u8)
          var k: i64=0
          while loc[k]!=(0 as u8) { urlbuf[k]=loc[k]; k=k+1 }
          urlbuf[k]=0 as u8
          hop=hop+1
      } else {
          if rc==2 { return 2 }
          if rc==1 { return 1 }
          mgo_put("probe rc=" as *u8); mgo_num(rc); mgo_put(" status=" as *u8); mgo_num(st[0]); mgo_put("\n" as *u8)
          return rc
      }
      lastrc=rc
  }
  return 0-8
172}
173
174func mg_receipt(verdict: *u8, dest: *u8, bytes: i64, hex: *u8, url: *u8) -> i64 {
  let fd: i64 = sys_openat_append(MG_RECEIPT_PATH, MODE_0644)
  if fd<0 { mgo_put("receipt=UNWRITTEN (knowledge/status/modelget.log not writable from this cwd)\n" as *u8); return 0-1 }
  let line: *u8 = sys_mmap(8192)
  var o: i64=0
  o = mg_cat(line, o, "ts=" as *u8)
  let t: *u8 = sys_mmap(28)
  var m: i64 = sys_now_realtime_sec()
  var k: i64=0
  if m==0 { t[0]=48 as u8; k=1 }
  while m>0 { t[k]=(48+(m%10)) as u8; m=m/10; k=k+1 }
  var j: i64=k-1
  while j>=0 { line[o]=t[j]; o=o+1; j=j-1 }
  o = mg_cat(line, o, " verdict=" as *u8)
  o = mg_cat(line, o, verdict)
  o = mg_cat(line, o, " dest=" as *u8)
  o = mg_cat(line, o, dest)
  o = mg_cat(line, o, " bytes=" as *u8)
  var b: i64 = bytes
  if b<0 { line[o]=45 as u8; o=o+1; b=0-b }
  k=0
  if b==0 { t[0]=48 as u8; k=1 }
  while b>0 { t[k]=(48+(b%10)) as u8; b=b/10; k=k+1 }
  j=k-1
  while j>=0 { line[o]=t[j]; o=o+1; j=j-1 }
  o = mg_cat(line, o, " sha256=" as *u8)
  o = mg_cat(line, o, hex)
  o = mg_cat(line, o, " url=" as *u8)
  o = mg_cat(line, o, url)
  line[o]=10 as u8
  o=o+1
  sys_write(fd, line, o)
  sys_fsync(fd)
  sys_close(fd)
  mgo_put("receipt=knowledge/status/modelget.log\n" as *u8)
  return 0
210}
211
212func main(argc: i64, argv: *i64) -> i64 {
  if argc<4 {
      mgo_put("usage: nx_model_get <url> <expected-sha256> <dest-name>\n  dest-name is relative to the model-store root (diffusion/... or loras/...)\n" as *u8)
      return 2
  }
  let url: *u8 = argv[1] as *u8
  let hexarg: *u8 = argv[2] as *u8
  let name: *u8 = argv[3] as *u8
220
  let exphex: *u8 = sys_mmap(80)
  if mg_hex_norm(hexarg, exphex)!=1 {
      mgo_put("REFUSED-BAD-HASH rule=expected-sha256-must-be-64-hex -- got " as *u8); mgo_put(hexarg); mgo_put("\n" as *u8)
      return 2
  }
  if mg_dest_ok(name)!=1 {
      mgo_put("REFUSED-DEST-NAME rule=relative-name-only (no leading slash, no .., no backslash) -- got " as *u8); mgo_put(name); mgo_put("\n" as *u8)
      return 2
  }
230
  // ---- model-store root + window from the conf (env > conf > code default: rule 17; provenance announced) ----
  let root: *u8 = sys_mmap(2048)
  var window: i64 = MG_WINDOW_DEFAULT
  var rootsrc: *u8 = "DEFAULT" as *u8
  let clen: *i64 = (sys_mmap(16)) as *i64
  let conf: *u8 = sys_read_file(MG_CONF_PATH, clen)
  if (conf as i64)!=0 {
      let rl: i64 = mg_conf_val(conf, clen[0], "root" as *u8, root, 2048)
      if rl>0 { rootsrc = "CONF" as *u8 }
      let wv: *u8 = sys_mmap(64)
      let wl: i64 = mg_conf_val(conf, clen[0], "window_bytes" as *u8, wv, 64)
      if wl>0 { let w: i64 = mg_parse_uint(wv, wl, 0); if w>0 { window = w } }
  }
  if root[0]==(0 as u8) { mg_cat(root, 0, MG_ROOT_DEFAULT); root[mg_slen(MG_ROOT_DEFAULT)]=0 as u8 }
  mgo_put("store_root=" as *u8); mgo_put(root); mgo_put(" (" as *u8); mgo_put(rootsrc); mgo_put(") window_bytes=" as *u8); mgo_num(window); mgo_put("\n" as *u8)
246
  let dest: *u8 = sys_mmap(4096)
  mg_join(root, name, dest)
  let part: *u8 = sys_mmap(4096)
  var po: i64 = mg_cat(part, 0, dest)
  po = mg_cat(part, po, ".part" as *u8)
  part[po]=0 as u8
253
  // ---- idempotency: an existing verified dest is a no-op (no network) ----
  let efd: i64 = sys_openat_rd(dest)
  if efd>=0 {
      sys_close(efd)
      let ehex: *u8 = sys_mmap(80)
      let en: i64 = mg_hash_file_hex(dest, ehex)
      if en>=0 { if wp_streq(ehex, exphex)==1 {
          mgo_put("OK-ALREADY-VERIFIED dest=" as *u8); mgo_put(dest); mgo_put(" bytes=" as *u8); mgo_num(en); mgo_put(" sha256=" as *u8); mgo_put(ehex); mgo_put("\n" as *u8)
          return 0
      } }
      mgo_put("REFUSED-EXISTING-MISMATCH dest=" as *u8); mgo_put(dest); mgo_put(" bytes=" as *u8); mgo_num(en); mgo_put(" sha256=" as *u8); mgo_put(ehex); mgo_put(" expected=" as *u8); mgo_put(exphex)
      mgo_put(" rule=additive-only -- an existing store artifact is never overwritten or deleted; remove it deliberately or pick a new dest-name\n" as *u8)
      return 1
  }
268
  // ---- trust store (same Mozilla bundle every sovereign fetcher loads) ----
  var tr: i64 = nx_trust_store_load_from_certdata("data/mozilla_certdata.txt" as *u8, MG_CERT_ROOTS, MG_CERT_CAP)
  if tr<=0 { tr = nx_trust_store_load_from_certdata("/tmp/mozilla_certdata.txt" as *u8, MG_CERT_ROOTS, MG_CERT_CAP) }
  if tr<=0 { mgo_put("UNPROVEN trust-store-load-failed (need data/mozilla_certdata.txt on cwd) -- nothing fetched\n" as *u8); return 3 }
  let store: *TrustStore = tr as *TrustStore
274
  return mg_fetch(url, exphex, dest, part, store, window)
276}
277
278// G9 mg_fetch -- THE CONTRACT SYMBOL (/compare/gen watch row MODEL: hash-verified acquisition). resolve redirects
279// -> range-window stream (hash as it lands) -> fail-closed verify over the FULL body -> rename into the store.
280// exit-code shaped return: 0 VERIFIED | 1 REFUSED-HASH-MISMATCH | 3 UNPROVEN (transport; partial kept for resume)
281func mg_fetch(url: *u8, exphex: *u8, dest: *u8, part: *u8, store: *TrustStore, window: i64) -> i64 {
  // ---- resolve redirects to the final range-answering url ----
  let urlbuf: *u8 = sys_mmap(MG_URLCAP)
  var ui: i64=0
  while url[ui]!=(0 as u8) { urlbuf[ui]=url[ui]; ui=ui+1 }
  urlbuf[ui]=0 as u8
  let chromebox: *i64 = (sys_mmap(16)) as *i64
  chromebox[0]=0
  let totbox: *i64 = (sys_mmap(16)) as *i64
  var mode: i64 = mg_resolve(urlbuf, store, chromebox, totbox)
  if mode<0 { mgo_put("UNPROVEN transport-resolve-failed rc=" as *u8); mgo_num(mode); mgo_put(" -- nothing stored\n" as *u8); return 3 }
  var total: i64 = totbox[0]
  if total<=0 { mgo_put("UNPROVEN no-declared-length (no Content-Range total, no Content-Length) -- refusing: completeness would be unjudgeable\n" as *u8); return 3 }
  mgo_put("resolved mode=" as *u8); mgo_num(mode); mgo_put(" total_bytes=" as *u8); mgo_num(total); mgo_put(" chrome_hello=" as *u8); mgo_num(chromebox[0]); mgo_put("\n" as *u8)
  if mode==1 { if total>window {
      mgo_put("REFUSED-NO-RANGE-SUPPORT rule=a-body-larger-than-the-window-must-stream -- server ignored Range and the body (" as *u8); mgo_num(total)
      mgo_put(" B) exceeds window_bytes (" as *u8); mgo_num(window); mgo_put(" B); refusing rather than truncate\n" as *u8)
      return 3
  } }
300
  // ---- streaming hash context; resume from .part if one exists ----
  let ctx_raw: *u8 = sys_mmap(256)
  let ctx: *Sha256 = ctx_raw as *Sha256
  sha256_init(ctx)
  var base: i64=0
  let pfd: i64 = sys_openat_rd(part)
  if pfd>=0 {
      sys_close(pfd)
      let pn: i64 = mg_hash_file_into(ctx, part)
      if pn>0 { if pn<total {
          base=pn
          mgo_put("RESUME part=" as *u8); mgo_put(part); mgo_put(" bytes_already=" as *u8); mgo_num(pn); mgo_put("\n" as *u8)
      } }
      if base==0 {
          sys_unlinkat(part)
          sha256_init(ctx)
          mgo_put("part-discarded (empty or not smaller than total) -- restarting from 0\n" as *u8)
      }
  }
320
  let afd: i64 = sys_openat_append(part, MODE_0644)
  if afd<0 { mgo_put("UNPROVEN cannot-open-part-for-append dest=" as *u8); mgo_put(part); mgo_put("\n" as *u8); return 3 }
323
  // ---- the window loop: fetch range -> hash -> append; never more than one window in memory ----
  let wbuf: *u8 = sys_mmap(window)
  let st: *i64 = (sys_mmap(16)) as *i64
  var off: i64=base
  var winfails: i64=0
  var wincount: i64=0
  var stalled: i64=0
  while off<total {
      var want: i64=window
      if off+want>total { want=total-off }
      let n: i64 = nx_https_fetch_range(urlbuf, store, wbuf, window, off, off+want-1, st, chromebox[0])
      var okwin: i64=0
      if n==want {
          if st[0]==206 { okwin=1 }
          if st[0]==200 { if off==0 { if want==total { okwin=1 } } }
      }
      if okwin==0 {
          winfails=winfails+1
          mgo_put("window-retry off=" as *u8); mgo_num(off); mgo_put(" got=" as *u8); mgo_num(n); mgo_put(" want=" as *u8); mgo_num(want); mgo_put(" status=" as *u8); mgo_num(st[0]); mgo_put(" fail=" as *u8); mgo_num(winfails); mgo_put("\n" as *u8)
          if winfails>=MG_WIN_RETRIES {
              sys_close(afd)
              mgo_put("UNPROVEN transport-stall at off=" as *u8); mgo_num(off); mgo_put(" of " as *u8); mgo_num(total)
              mgo_put(" -- PARTIAL KEPT for resume at " as *u8); mgo_put(part); mgo_put(" (re-run the same command)\n" as *u8)
              off=total
              stalled=1
          } else {
              // a signed CDN url may have expired mid-pull: re-resolve from the ORIGINAL url and retry this window
              var ri: i64=0
              while url[ri]!=(0 as u8) { urlbuf[ri]=url[ri]; ri=ri+1 }
              urlbuf[ri]=0 as u8
              let m2: i64 = mg_resolve(urlbuf, store, chromebox, totbox)
              if m2<0 { mgo_put("re-resolve failed rc=" as *u8); mgo_num(m2); mgo_put("\n" as *u8) }
          }
      } else {
          winfails=0
          sha256_update(ctx, wbuf, n)
          var sent: i64=0
          var wbad: i64=0
          while sent<n { let w: i64 = sys_write(afd, (wbuf as i64 + sent) as *u8, n-sent); if w<=0 { wbad=1; sent=n } else { sent=sent+w } }
          if wbad==1 {
              sys_close(afd)
              mgo_put("UNPROVEN disk-write-failed at off=" as *u8); mgo_num(off); mgo_put(" -- PARTIAL KEPT at " as *u8); mgo_put(part); mgo_put("\n" as *u8)
              off=total
              stalled=1
          } else {
              off=off+n
              wincount=wincount+1
              if wincount%MG_PROGRESS_EVERY==0 { mgo_put("progress bytes=" as *u8); mgo_num(off); mgo_put(" of " as *u8); mgo_num(total); mgo_put("\n" as *u8) }
          }
      }
  }
  if stalled==1 { return 3 }
  sys_fsync(afd)
  sys_close(afd)
378
  // ---- fail-closed verify over the FULL body (the nx_weights_pull law: never store a corrupt artifact) ----
  let dig: *u8 = sys_mmap(32)
  sha256_final(ctx, dig)
  let got: *u8 = sys_mmap(80)
  wp_hexenc(dig, 32, got)
  if wp_streq(got, exphex)==1 {
      sys_renameat(part, dest)
      mgo_put("VERIFIED dest=" as *u8); mgo_put(dest); mgo_put(" bytes=" as *u8); mgo_num(total); mgo_put(" sha256=" as *u8); mgo_put(got); mgo_put("\n" as *u8)
      mg_receipt("VERIFIED" as *u8, dest, total, got, url)
      return 0
  }
  sys_unlinkat(part)
  mgo_put("REFUSED-HASH-MISMATCH expected=" as *u8); mgo_put(exphex); mgo_put(" got=" as *u8); mgo_put(got); mgo_put(" bytes=" as *u8); mgo_num(total)
  mgo_put(" rule=fail-closed-verify -- the partial was DELETED; a corrupt, tampered or truncated body is never stored (nx_weights_pull law)\n" as *u8)
  mg_receipt("REFUSED-HASH-MISMATCH" as *u8, dest, total, got, url)
  return 1
395}
nx_model_get.nx source

`nx_model_get.nx` source
