Security hardening (OWASP Top 10 review)

src/native-tools/bridge.py

@@ -80,6 +80,18 @@ def usb_gone(e):
         e = e.__cause__ or e.__context__
     return False
 
+class TooLarge(RuntimeError):
+    """A request body or zip import over its size cap — HTTP 413."""
+
+
+# Size caps (A10: a huge body or a zip bomb would otherwise exhaust memory or
+# fill the disk). Generous vs real use: the device's whole sample pool is
+# 14.6 MB, a full backup is ~55 files — but a WAV upload arrives BEFORE
+# conversion, so a long 24-bit/96 kHz source can be several times that.
+MAX_BODY = 256 << 20                  # any request body
+MAX_ZIP_BYTES = 64 << 20              # a backup zip, uncompressed total
+MAX_ZIP_FILES = 1000                  # … and member count
+
 VERSION = '1.17.3'   # current app version; kept in sync by tools/stamp-docs-version.mjs
 # static root; MSMPL_WEB_ROOT overrides for the bundled app (PyInstaller datas)
 WEB_ROOT = os.environ.get('MSMPL_WEB_ROOT') or \
@@ -1492,6 +1504,7 @@ def import_backup_zip(data):
     """Unpack an uploaded backup .zip into a fresh BACKUP_ROOT/<name>.
     Zip-slip safe (every member must stay inside the target) and validated
     (must carry a manifest.json). Returns the new directory name."""
+    import shutil
     import zipfile
     _check_backup_writable()
     os.makedirs(BACKUP_ROOT, exist_ok=True)
@@ -1500,9 +1513,16 @@ def import_backup_zip(data):
     except zipfile.BadZipFile:
         raise RuntimeError('not a zip file') from None   # client input → HTTP 400
     with zf as z:
-        names = [n for n in z.namelist() if not n.endswith('/')]
+        infos = [i for i in z.infolist() if not i.filename.endswith('/')]
+        names = [i.filename for i in infos]
         if not any(n.endswith('manifest.json') for n in names):
             raise RuntimeError('not a backup zip (no manifest.json)')
+        # zip-bomb guard, up front from the declared sizes (cheap) — the copy
+        # loop below re-counts the ACTUAL bytes, so a lying header can't win
+        if len(infos) > MAX_ZIP_FILES or \
+                sum(i.file_size for i in infos) > MAX_ZIP_BYTES:
+            raise TooLarge('backup zip too large (over %d MB or %d files '
+                           'unpacked)' % (MAX_ZIP_BYTES >> 20, MAX_ZIP_FILES))
         # derive a name: the zip's common top folder, else a timestamp
         tops = {n.split('/', 1)[0] for n in names if '/' in n}
         base = tops.pop() if len(tops) == 1 else 'import'
@@ -1511,14 +1531,27 @@ def import_backup_zip(data):
         while os.path.exists(dest):
             dest = backup_dir('%s-%d' % (base, n)); n += 1
         root = os.path.realpath(dest) + os.sep
-        for member in names:
-            out = os.path.realpath(os.path.join(dest, os.path.relpath(
-                member, base) if member.startswith(base + '/') else member))
-            if not out.startswith(root):
-                raise RuntimeError('unsafe path in zip: %r' % member)
-            os.makedirs(os.path.dirname(out), exist_ok=True)
-            with z.open(member) as fsrc, open(out, 'wb') as fdst:
-                fdst.write(fsrc.read())
+        total = 0
+        try:
+            for member in names:
+                out = os.path.realpath(os.path.join(dest, os.path.relpath(
+                    member, base) if member.startswith(base + '/') else member))
+                if not out.startswith(root):
+                    raise RuntimeError('unsafe path in zip: %r' % member)
+                os.makedirs(os.path.dirname(out), exist_ok=True)
+                with z.open(member) as fsrc, open(out, 'wb') as fdst:
+                    for chunk in iter(lambda: fsrc.read(1 << 20), b''):
+                        total += len(chunk)
+                        if total > MAX_ZIP_BYTES:
+                            raise TooLarge('backup zip too large (over %d MB '
+                                           'unpacked)' % (MAX_ZIP_BYTES >> 20))
+                        fdst.write(chunk)
+        except zipfile.BadZipFile as e:   # e.g. CRC mismatch (a size-lying
+            shutil.rmtree(dest, ignore_errors=True)   # header stops short)
+            raise RuntimeError('corrupt zip file (%s)' % e) from None  # → 400
+        except BaseException:
+            shutil.rmtree(dest, ignore_errors=True)   # no half-imported backup
+            raise
     return os.path.basename(dest)
 
 
@@ -1603,6 +1636,16 @@ class Handler(BaseHTTPRequestHandler):
     protocol_version = 'HTTP/1.1'
 
     # -- helpers --------------------------------------------------------------
+    def end_headers(self):
+        """Hardening headers on EVERY response (JSON, static, SSE, errors):
+        no framing by another site (clickjacking — a hidden iframe of the
+        editor could trick clicks onto RESTORE/confirm), and no MIME sniffing
+        (MIME covers every file type the app ships)."""
+        self.send_header('X-Frame-Options', 'DENY')
+        self.send_header('Content-Security-Policy', "frame-ancestors 'none'")
+        self.send_header('X-Content-Type-Options', 'nosniff')
+        super().end_headers()
+
     def _json(self, obj, code=200):
         body = json.dumps(obj).encode()
         self.send_response(code)
@@ -1632,8 +1675,12 @@ def _fail(self, e):
         self._err(e)
 
     def _body(self):
-        """The raw request body (Content-Length bytes)."""
-        return self.rfile.read(int(self.headers.get('Content-Length', 0)))
+        """The raw request body (Content-Length bytes), capped at MAX_BODY."""
+        n = int(self.headers.get('Content-Length', 0))
+        if not 0 <= n <= MAX_BODY:
+            raise TooLarge('request body too large (max %d MB)'
+                           % (MAX_BODY >> 20))
+        return self.rfile.read(n)
 
     @staticmethod
     def _slot(s):
@@ -1672,6 +1719,26 @@ def _bad_host(self):
         self._err('forbidden host', 403)
         return True
 
+    def _cross_site(self):
+        """Reject state-changing requests sent by ANOTHER site. The Host check
+        stops DNS rebinding, but not CSRF: any page can fire a no-cors POST at
+        http://localhost:<port> (text/plain body, response unreadable) and the
+        side effect still lands — e.g. import its own backup zip, then restore
+        it over a user bank in flash. Browsers always send Origin on a POST
+        (it must be exactly our own origin — another localhost port is
+        another origin) and mark cross-site fetches via Sec-Fetch-Site.
+        Non-browser clients (menu-bar app, CLI, tests) send neither and pass.
+        Returns True (and answers 403) when the request is cross-site."""
+        origin = self.headers.get('Origin')
+        if origin is not None:
+            ok = origin == 'http://%s' % (self.headers.get('Host') or '').strip()
+        else:
+            ok = self.headers.get('Sec-Fetch-Site') != 'cross-site'
+        if ok:
+            return False
+        self._err('cross-site request refused', 403)
+        return True
+
     # -- routing ---------------------------------------------------------------
     def do_GET(self):
         if self._bad_host():
@@ -1745,7 +1812,7 @@ def do_GET(self):
             self._fail(e)
 
     def do_POST(self):
-        if self._bad_host():
+        if self._bad_host() or self._cross_site():
             return
         path, _, query = self.path.partition('?')
         # the app percent-encodes query values (encodeURIComponent) — decode,
@@ -1927,6 +1994,9 @@ def do_POST(self):
             pass                              # client went away mid-request
         except DeviceGone as e:
             self._err(e, 503)                 # bridge up, device unclaimed
+        except TooLarge as e:
+            self.close_connection = True      # the unread body would desync
+            self._err(e, 413)                 # the keep-alive stream
         except KeyError as e:                 # missing body field
             self._err('missing field %s' % e, 400)
         # client-input errors are 400s, not tracebacks: bad JSON / non-numeric

src/native-tools/test_bridge.py

@@ -31,6 +31,43 @@ def req(method, path, body=None, headers=None):
 assert st == 200 and status['mock'] is True
 assert status['shutdown'] is False and status['daemon'] is False
 
+# --- request-origin guards (DNS rebinding + cross-site request forgery) ---------
+_vol = json.dumps({'value': 100})
+st, _, _ = req('GET', '/api/status', headers={'Host': 'evil.example'})
+assert st == 403                                # rebinding: foreign Host
+for hdrs in ({'Origin': 'https://evil.example'},            # another site
+             {'Origin': 'http://localhost:3000'},           # another local port
+             {'Origin': 'null'},                            # sandboxed/opaque
+             {'Sec-Fetch-Site': 'cross-site'}):             # no Origin, flagged
+    st, _, data = req('POST', '/api/master-volume', _vol, headers=hdrs)
+    assert st == 403 and b'cross-site' in data, (hdrs, st)
+for hdrs in ({'Origin': 'http://127.0.0.1:%d' % PORT,      # the app itself
+              'Sec-Fetch-Site': 'same-origin'},
+             {}):                                           # menu-bar app / CLI
+    st, _, _ = req('POST', '/api/master-volume', _vol, headers=hdrs)
+    assert st == 200, (hdrs, st)
+print('origin guards: OK')
+
+# --- hardening headers on every response kind (static, JSON, error) ------------
+for _path in ('/app.html', '/api/status', '/api/nope'):
+    _c = http.client.HTTPConnection('127.0.0.1', PORT, timeout=10)
+    _c.request('GET', _path)
+    _r = _c.getresponse(); _r.read(); _c.close()
+    assert _r.getheader('X-Frame-Options') == 'DENY', _path
+    assert _r.getheader('Content-Security-Policy') == "frame-ancestors 'none'", _path
+    assert _r.getheader('X-Content-Type-Options') == 'nosniff', _path
+print('security headers: OK')
+
+# --- request-body cap → 413 (and the connection is closed, body unread) ---------
+_saved_max = B.MAX_BODY
+B.MAX_BODY = 16
+st, _, data = req('POST', '/api/master-volume', b'{"value": 100}   ')   # 17 B
+assert st == 413 and b'too large' in data, (st, data)
+st, _, _ = req('POST', '/api/master-volume', _vol)                     # 15 B
+assert st == 200
+B.MAX_BODY = _saved_max
+print('body size cap: OK')
+
 # --- daemon-mode primitives (device release + lazy-claim throttle) --------------
 st, _, data = req('POST', '/api/release')       # nothing claimed → safe no-op
 assert st == 200 and json.loads(data) == {'released': True, 'was_claimed': False}
@@ -656,6 +693,44 @@ def send_sysex(self, midi, cable=None):
     B.import_backup_zip(bad.getvalue()); raise AssertionError('accepted non-backup')
 except RuntimeError as e:
     assert 'no manifest' in str(e)
+# zip-bomb caps: too many bytes / files → TooLarge, and NO half-imported dir
+def _zip_of(files):
+    b = io.BytesIO()
+    with _zf.ZipFile(b, 'w', _zf.ZIP_DEFLATED) as z:
+        for n, d in files.items():
+            z.writestr(n, d)
+    return b.getvalue()
+_bomb = _zip_of({'bomb/manifest.json': '{}', 'bomb/x.bin': b'\0' * (2 << 20)})
+assert len(_bomb) < 64 << 10                     # tiny on the wire, 2 MB unpacked
+_saved_caps = B.MAX_ZIP_BYTES, B.MAX_ZIP_FILES
+for _caps, _z in (((1 << 20, 1000), _bomb),
+                  ((64 << 20, 3), _zip_of({'many/manifest.json': '{}',
+                                           **{'many/%d' % i: 'x' for i in range(5)}}))):
+    B.MAX_ZIP_BYTES, B.MAX_ZIP_FILES = _caps
+    _before = sorted(os.listdir(B.BACKUP_ROOT))
+    try:
+        B.import_backup_zip(_z); raise AssertionError('accepted oversized zip')
+    except B.TooLarge as e:
+        assert 'too large' in str(e)
+    assert sorted(os.listdir(B.BACKUP_ROOT)) == _before    # nothing left behind
+B.MAX_ZIP_BYTES, B.MAX_ZIP_FILES = _saved_caps
+# a header that UNDER-declares its size: zipfile stops at the declared size,
+# then fails the CRC → a clean client error (400), again with nothing left
+B.MAX_ZIP_BYTES = 1 << 20
+_lie = bytearray(_bomb)
+for _i in range(len(_lie) - 4):                 # patch every file_size field of
+    if _lie[_i:_i + 4] in (b'PK\x03\x04', b'PK\x01\x02'):   # x.bin to 1 byte
+        _off = 22 if _lie[_i:_i + 4] == b'PK\x03\x04' else 24
+        if int.from_bytes(_lie[_i + _off:_i + _off + 4], 'little') == 2 << 20:
+            _lie[_i + _off:_i + _off + 4] = (1).to_bytes(4, 'little')
+_before = sorted(os.listdir(B.BACKUP_ROOT))
+try:
+    B.import_backup_zip(bytes(_lie)); raise AssertionError('lying zip accepted')
+except RuntimeError as e:
+    assert 'corrupt zip' in str(e), e
+assert sorted(os.listdir(B.BACKUP_ROOT)) == _before
+B.MAX_ZIP_BYTES = _saved_caps[0]
+print('zip import caps: OK')
 try:
     B.backup_zip('../etc'); raise AssertionError('accepted traversal')
 except RuntimeError as e:

src/native-tools/test_daemon_dirs.py

@@ -0,0 +1,62 @@
+"""Offline test of the Editor app daemon's backup-folder guard
+(tools/bundle/editor_daemon.secure_dir): the ROOT daemon must never follow a
+symlink or use a folder another account owns or can write. Runs unprivileged:
+`owner` is the current uid here (root in the real daemon)."""
+import os
+import stat
+import sys
+import tempfile
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.join(HERE, '..', '..', 'tools', 'bundle'))
+import editor_daemon as D  # noqa: E402
+
+me = os.getuid()
+
+
+def fresh_base():
+    base = tempfile.mkdtemp()
+    os.chmod(base, 0o1777)                 # like /Users/Shared: shared + sticky
+    return base
+
+
+# fresh install: the whole chain is created, owned by us, 0755
+base = fresh_base()
+path = os.path.join(base, 'DehliMusikk', 'microSAMPLER Editor', 'backups')
+assert D.secure_dir(path, base, owner=me)
+for sub in ('DehliMusikk', 'DehliMusikk/microSAMPLER Editor', path):
+    st = os.lstat(os.path.join(base, sub))
+    assert stat.S_ISDIR(st.st_mode) and stat.S_IMODE(st.st_mode) == 0o755, sub
+assert D.secure_dir(path, base, owner=me)     # existing + intact: still fine
+
+# a pre-planted symlink at any level is refused, and NOT followed
+base = fresh_base()
+target = tempfile.mkdtemp()
+os.symlink(target, os.path.join(base, 'DehliMusikk'))
+assert not D.secure_dir(os.path.join(base, 'DehliMusikk', 'x'), base, owner=me)
+assert os.listdir(target) == []                # nothing created through it
+
+# a component owned by someone else is refused
+base = fresh_base()
+assert not D.secure_dir(os.path.join(base, 'a', 'b'), base, owner=me + 1)
+
+# a group/other-writable component below base is refused
+base = fresh_base()
+os.mkdir(os.path.join(base, 'DehliMusikk'))
+os.chmod(os.path.join(base, 'DehliMusikk'), 0o777)
+assert not D.secure_dir(os.path.join(base, 'DehliMusikk', 'x'), base, owner=me)
+
+# a regular FILE where a folder should be is refused
+base = fresh_base()
+open(os.path.join(base, 'DehliMusikk'), 'w').close()
+assert not D.secure_dir(os.path.join(base, 'DehliMusikk', 'x'), base, owner=me)
+
+# a world-writable base WITHOUT the sticky bit is refused (entries replaceable)
+base = fresh_base()
+os.chmod(base, 0o777)
+assert not D.secure_dir(os.path.join(base, 'x'), base, owner=me)
+
+# a path outside base is refused
+assert not D.secure_dir('/etc', fresh_base(), owner=me)
+
+print('daemon backup-folder guard: OK')