Verify generated native SDK archive contents

tools/migration_ci.py

@@ -19,9 +19,9 @@ def digest(path):
 
 
 def native_input_paths(project, expected=None):
-    from native_runtime import SERVER_INPUTS
+    from native_runtime import verify_server_inputs
     native=project/'build/native-server-distribution'
-    expected=SERVER_INPUTS if expected is None else expected
+    expected=verify_server_inputs(native) if expected is None else expected
     if not expected:raise ValueError('Missing native SDK manifest')
     paths=[]
     for name,sha in sorted(expected.items()):

tools/native117_generated_sdk.json

@@ -0,0 +1,14 @@
+{
+  "libraries/net/minecraft/server/1.17.1-20210706.113038/server-1.17.1-20210706.113038-srg.jar": {
+    "entries": 4677,
+    "content_and_metadata_sha256": "38885B270B4072A6F4C61764FEF61C2839C00505E58C7A193F5C896EB3834F7D",
+    "timestamp_policy": "installer-times",
+    "canonical_archive_sha256": "A00E5004F190A0F221D8001B2E04609CE744A129F7764181AD10FBEFB1237EAA"
+  },
+  "libraries/net/minecraftforge/forge/1.17.1-37.1.1/forge-1.17.1-37.1.1-server.jar": {
+    "entries": 1252,
+    "content_and_metadata_sha256": "AF6A206D4398A373937D85C240CD16ACC2833A9A8493F483FA31E95870DFF3F0",
+    "timestamp_policy": "exact",
+    "canonical_archive_sha256": "35DB91BCD011CAD8B812BA165CD7F5AD934CEEEE9050621423F9E5D227D7F895"
+  }
+}

tools/native_runtime.py

@@ -39,6 +39,29 @@ def fingerprint(directory):
             for p in sorted(directory.rglob('*')) if p.is_file()}
 
 
+def verify_server_inputs(distribution):
+    """Pin raw inputs and seal both installer-generated archive contents.
+
+    Only installer directory timestamps and ZIP entry order may vary. The
+    returned hashes bind the actual SDK bytes prepared for every process.
+    """
+    from native_sdk_contents import verify_generated_archive
+    generated=json.loads(Path(__file__).with_name('native117_generated_sdk.json').read_text())
+    actual={}
+    for name,expected in SERVER_INPUTS.items():
+        artifact=Path(distribution)/name
+        if artifact.is_symlink() or not artifact.is_file():raise FileNotFoundError('Missing native37 SDK input: '+name)
+        value=digest(artifact)
+        if value!=expected:
+            seal=generated.get(name)
+            if not seal or seal['canonical_archive_sha256']!=expected:
+                raise RuntimeError('Incorrect official Forge37 server input: '+name)
+            verify_generated_archive(artifact,seal)
+            if digest(artifact)!=value:raise RuntimeError('Native SDK changed during verification: '+name)
+        actual[name]=value
+    return actual
+
+
 def seed_loader_defaults(home, deterministic=False, include_default_path=True):
     """Seed an owned home; disable optional HTTP checks only for probes."""
     config=home/'config/fml.toml'
@@ -57,10 +80,7 @@ def prepare_server(home, distribution, candidate=None, probe=None):
     libraries = distribution / 'libraries'
     if not libraries.is_dir():
         raise FileNotFoundError('Incomplete official Forge 37 server installation')
-    for name, expected in SERVER_INPUTS.items():
-        artifact = distribution / name
-        if not artifact.is_file() or digest(artifact) != expected:
-            raise RuntimeError('Incorrect official Forge 37 server input: ' + name)
+    verify_server_inputs(distribution)
     home.mkdir(parents=True, exist_ok=False)
     (home / 'mods').mkdir()
     seed_loader_defaults(home, deterministic=probe is not None)
@@ -279,8 +299,7 @@ def install(java, home, client=False):
         from native117_client import prepare
         inputs=prepare(home,home/'prepared-inputs')['files']
     else:
-        inputs={name:digest(home/name) for name in SERVER_INPUTS}
-        if inputs!=SERVER_INPUTS:raise RuntimeError('Official Forge37 server installation checksum mismatch')
+        inputs=verify_server_inputs(home)
     (home/'runtime-inputs.json').write_text(json.dumps({'installer_sha256':INSTALLER_SHA256,'client':client,'files':inputs},indent=2)+'\n')
     return home
 

tools/native_sdk_contents.py

@@ -0,0 +1,37 @@
+"""Verify generated native SDK archives despite installer container metadata."""
+import hashlib,json,struct,zipfile
+from pathlib import PurePosixPath
+
+def timestamp_extra(extra):
+    # Java's directory-backed installer emits NTFS times on Windows and
+    # extended Unix times on Linux. No other extra field is admitted here.
+    if len(extra)<4:raise ValueError('Missing installer timestamp metadata')
+    kind,size=struct.unpack('<HH',extra[:4]);data=extra[4:]
+    if size!=len(data):raise ValueError('Truncated installer timestamp metadata')
+    if kind==0x000a:
+        if len(data)!=32 or data[:8]!=b'\0\0\0\0\1\0\x18\0':raise ValueError('Invalid NTFS timestamp metadata')
+    elif kind==0x5455:
+        if len(data)!=5 or data[0]!=7:raise ValueError('Invalid Unix timestamp metadata')
+    else:raise ValueError('Unknown generated SDK extra field')
+
+def archive_seal(path, *, timestamp_policy):
+    if timestamp_policy not in ('exact','installer-times'):raise ValueError('Unknown SDK archive policy')
+    with zipfile.ZipFile(path) as archive:
+        entries=archive.infolist();names=[i.filename for i in entries]
+        if not names or len(names)!=len(set(names)):raise ValueError('Missing or duplicate SDK entries')
+        result=[]
+        for entry in sorted(entries,key=lambda e:e.filename):
+            name=PurePosixPath(entry.filename)
+            if name.is_absolute() or '..' in name.parts or '\\' in entry.filename or ':' in entry.filename:raise ValueError('Unsafe SDK archive entry')
+            if timestamp_policy=='installer-times':timestamp_extra(entry.extra)
+            metadata={key:getattr(entry,key) for key in ('compress_type','create_system','create_version','extract_version','flag_bits','volume','internal_attr','external_attr','CRC','compress_size','file_size')}
+            metadata.update(name=entry.filename,comment=entry.comment.hex(),content_sha256=hashlib.sha256(archive.read(entry)).hexdigest().upper(),date_time=None if timestamp_policy=='installer-times' and entry.is_dir() else entry.date_time,extra='installer-times' if timestamp_policy=='installer-times' else entry.extra.hex())
+            result.append(metadata)
+        value={'comment':archive.comment.hex(),'entries':result}
+    return {'entries':len(entries),'content_and_metadata_sha256':hashlib.sha256(json.dumps(value,sort_keys=True,separators=(',',':')).encode()).hexdigest().upper()}
+
+def verify_generated_archive(path, expected):
+    actual=archive_seal(path,timestamp_policy=expected['timestamp_policy'])
+    if actual!={key:expected[key] for key in ('entries','content_and_metadata_sha256')}:
+        raise ValueError('Changed generated native SDK class, resource or invariant metadata: '+str(path))
+    return actual

tools/test_native_sdk_contents.py

@@ -0,0 +1,46 @@
+import tempfile,unittest,zipfile,struct
+from pathlib import Path
+from native_sdk_contents import archive_seal,verify_generated_archive
+
+class GeneratedSDKTests(unittest.TestCase):
+    def jar(self,path,*,linux=False,change=None):
+        entries=[('pkg/',b''),('pkg/A.class',b'original class'),('pack.mcmeta',b'original resource')]
+        if linux:entries.reverse()
+        if change:entries=change(entries)
+        with zipfile.ZipFile(path,'x') as archive:
+            for name,data in entries:
+                item=zipfile.ZipInfo(name,date_time=(2020,1,1,0,0,0))
+                item.extra=struct.pack('<HH',0x5455,5)+b'\7'+struct.pack('<I',12345) if linux else struct.pack('<HH',0x000a,32)+b'\0\0\0\0\1\0\x18\0'+b'\0'*24
+                if linux and name.endswith('/'):item.date_time=(2026,10,30,11,14,36)
+                archive.writestr(item,data)
+    def expected(self,path):return dict(archive_seal(path,timestamp_policy='installer-times'),timestamp_policy='installer-times')
+    def test_only_directory_times_timestamp_extras_and_order_can_differ(self):
+        with tempfile.TemporaryDirectory() as d:
+            a,b=Path(d)/'a.jar',Path(d)/'b.jar';self.jar(a);self.jar(b,linux=True)
+            self.assertEqual(self.expected(a),self.expected(b));verify_generated_archive(b,self.expected(a))
+    def test_changed_missing_extra_or_truncated_contents_are_fatal(self):
+        changes=[lambda rows:[(n,b'changed') if n.endswith('.class') else (n,v) for n,v in rows],lambda rows:rows[:-1],lambda rows:rows+[('extra.class',b'extra')],lambda rows:[(n,v[:-1]) if n=='pack.mcmeta' else (n,v) for n,v in rows]]
+        for change in changes:
+            with self.subTest(change=change),tempfile.TemporaryDirectory() as d:
+                a,b=Path(d)/'a.jar',Path(d)/'b.jar';self.jar(a);self.jar(b,linux=True,change=change)
+                with self.assertRaises(ValueError):verify_generated_archive(b,self.expected(a))
+    def test_unknown_or_truncated_extra_fields_are_fatal(self):
+        from native_sdk_contents import timestamp_extra
+        for extra in (b'',b'UT\x05\0\7',b'XX\x01\0\0',b'UT\x05\0\0'+b'\0'*4):
+            with self.subTest(extra=extra),self.assertRaises(ValueError):timestamp_extra(extra)
+    def test_exact_metadata_policy_remains_strict(self):
+        with tempfile.TemporaryDirectory() as d:
+            a,b=Path(d)/'a.jar',Path(d)/'b.jar';self.jar(a);self.jar(b,linux=True)
+            self.assertNotEqual(archive_seal(a,timestamp_policy='exact'),archive_seal(b,timestamp_policy='exact'))
+    def test_changed_file_timestamp_and_duplicate_entries_are_fatal(self):
+        with tempfile.TemporaryDirectory() as d:
+            a,b=Path(d)/'a.jar',Path(d)/'b.jar';self.jar(a)
+            with zipfile.ZipFile(a) as left,zipfile.ZipFile(b,'x') as right:
+                for entry in left.infolist():
+                    if entry.filename.endswith('.class'):entry.date_time=(2021,1,1,0,0,0)
+                    right.writestr(entry,left.read(entry))
+            with self.assertRaises(ValueError):verify_generated_archive(b,self.expected(a))
+            with zipfile.ZipFile(b,'a') as right:right.writestr('pkg/A.class',b'duplicate')
+            with self.assertRaises(ValueError):verify_generated_archive(b,self.expected(a))
+
+if __name__=='__main__':unittest.main()