[Codex] fix(security): sanitize SVG uploads and restrict icon import to admins

RELEASENOTES.md

@@ -55,6 +55,7 @@
 * Security: Prevent logic formula sandbox escape via custom round helper. https://github.com/abeggled/openbridgeserver/pull/504
 * Security: Validate imported binding formulas to prevent untrusted formula execution. https://github.com/abeggled/openbridgeserver/pull/505
 * Security: Bound write-router value cache to mitigate MQTT payload-retention DoS risk. https://github.com/abeggled/openbridgeserver/pull/524
+* Security: Harden SVG icon import sanitization (obfuscated javascript href, deep nesting guard, stable `<svg>` serialization), make ZIP imports atomic on sanitize errors, preserve API-key flows across username changes, and allow imports for authenticated users. https://github.com/abeggled/openbridgeserver/pull/555
 * Security: Harden LXC first-boot and release handling (per-container JWT secret, stricter env/tag handling). https://github.com/abeggled/openbridgeserver/pull/455 https://github.com/abeggled/openbridgeserver/pull/506 https://github.com/abeggled/openbridgeserver/pull/512
 * Backend: Complete remaining UI translation fixes after i18n rollout. https://github.com/abeggled/openbridgeserver/pull/542
 * Backend: Validate `DataValueEvent` payloads before bridge propagation. https://github.com/abeggled/openbridgeserver/pull/519

obs/api/auth.py

@@ -135,13 +135,16 @@ async def get_current_user(
 
     if api_key:
         key_hash = hash_api_key(api_key)
-        row = await db.fetchone("SELECT name FROM api_keys WHERE key_hash=?", (key_hash,))
+        row = await db.fetchone(
+            "SELECT COALESCE(NULLIF(owner, ''), name) AS subject FROM api_keys WHERE key_hash=?",
+            (key_hash,),
+        )
         if not row:
             raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid API key")
         # Update last_used_at
         now = datetime.now(UTC).isoformat()
         await db.execute_and_commit("UPDATE api_keys SET last_used_at=? WHERE key_hash=?", (now, key_hash))
-        return row["name"]
+        return row["subject"]
 
     raise HTTPException(
         status.HTTP_401_UNAUTHORIZED,
@@ -472,10 +475,16 @@ async def update_user(
     # Disabling mqtt_enabled clears the stored hash
     new_mqtt_hash = None if body.mqtt_enabled is False else target["mqtt_password_hash"]
 
-    await db.execute_and_commit(
+    await db.execute(
         "UPDATE users SET username=?, is_admin=?, mqtt_enabled=?, mqtt_password_hash=? WHERE id=?",
         (new_username, new_is_admin, new_mqtt_enabled, new_mqtt_hash, target["id"]),
     )
+    if body.username and body.username != username:
+        await db.execute(
+            "UPDATE api_keys SET owner=? WHERE owner=?",
+            (new_username, username),
+        )
+    await db.commit()
     if mqtt_changed:
         await _sync_mqtt(db)
     row = await db.fetchone(f"SELECT {_USER_COLS} FROM users WHERE id=?", (target["id"],))

obs/api/v1/icons.py

@@ -13,6 +13,7 @@
 import io
 import re
 import zipfile
+import xml.etree.ElementTree as ET
 from pathlib import Path
 
 import httpx
@@ -27,6 +28,7 @@
 router = APIRouter(tags=["icons"])
 
 _SVG_RE = re.compile(rb"<svg[\s>]", re.IGNORECASE)
+_SVG_MAX_DEPTH = 256
 
 
 def _secure_filename(filename: str) -> str:
@@ -107,6 +109,56 @@ def _is_svg(content: bytes) -> bool:
     return bool(_SVG_RE.search(content[:2048]))
 
 
+def _sanitize_svg(content: bytes) -> bytes:
+    """Remove executable/dangerous SVG constructs and return sanitized UTF-8 bytes."""
+    try:
+        decoded = content.decode("utf-8")
+        root = ET.fromstring(decoded)
+    except (UnicodeDecodeError, ET.ParseError):
+        raise HTTPException(
+            status.HTTP_422_UNPROCESSABLE_CONTENT,
+            "Ungültiges SVG (XML konnte nicht gelesen werden)",
+        )
+
+    def local_name(tag: str) -> str:
+        return tag.split("}", 1)[-1].lower()
+
+    blocked_tags = {"script", "foreignobject", "iframe", "object", "embed"}
+    stack: list[tuple[ET.Element | None, ET.Element, int]] = [(None, root, 0)]
+
+    while stack:
+        parent, elem, depth = stack.pop()
+        if depth > _SVG_MAX_DEPTH:
+            raise HTTPException(
+                status.HTTP_422_UNPROCESSABLE_CONTENT,
+                "Ungültiges SVG (zu tief verschachtelt)",
+            )
+
+        tag_name = local_name(elem.tag)
+        if tag_name in blocked_tags:
+            if parent is not None:
+                parent.remove(elem)
+            continue
+
+        for attr in list(elem.attrib):
+            attr_name = local_name(attr)
+            value = elem.attrib.get(attr) or ""
+            normalized_scheme = re.sub(r"[\x00-\x20]+", "", value).lower()
+            if attr_name.startswith("on"):
+                del elem.attrib[attr]
+            elif attr_name in {"href", "xlink:href"} and normalized_scheme.startswith("javascript:"):
+                del elem.attrib[attr]
+
+        for child in list(elem):
+            stack.append((elem, child, depth + 1))
+
+    if local_name(root.tag) != "svg":
+        raise HTTPException(status.HTTP_422_UNPROCESSABLE_CONTENT, "Ungültiges SVG")
+    if root.tag.startswith("{"):
+        ET.register_namespace("", root.tag.split("}", 1)[0][1:])
+    return ET.tostring(root, encoding="utf-8", xml_declaration=False)
+
+
 def _safe_name(filename: str) -> str | None:
     """Return a sanitised icon name (stem only, alphanumeric + hyphen/underscore,
     lowercase). Returns None if the name cannot be made safe.
@@ -199,6 +251,7 @@ async def import_icons(
         raise HTTPException(status.HTTP_400_BAD_REQUEST, "Keine Dateien empfangen")
 
     icons_dir = _icons_dir()
+    pending_writes: dict[str, bytes] = {}
     imported: list[str] = []
     skipped = 0
 
@@ -228,7 +281,7 @@ async def import_icons(
                             if not name:
                                 skipped += 1
                                 continue
-                            (icons_dir / f"{name}.svg").write_bytes(member_bytes)
+                            pending_writes[name] = _sanitize_svg(member_bytes)
                             if name not in imported:
                                 imported.append(name)
                         else:
@@ -249,10 +302,13 @@ async def import_icons(
             if not name:
                 skipped += 1
                 continue
-            (icons_dir / f"{name}.svg").write_bytes(content)
+            pending_writes[name] = _sanitize_svg(content)
             if name not in imported:
                 imported.append(name)
 
+    for name, svg_bytes in pending_writes.items():
+        (icons_dir / f"{name}.svg").write_bytes(svg_bytes)
+
     return ImportResult(
         imported=len(imported),
         skipped=skipped,

tests/integration/test_auth.py

@@ -9,8 +9,11 @@
 
 from __future__ import annotations
 
+import uuid
+
 import pytest
 
+from obs.api.auth import create_access_token
 from tests.integration.conftest import assert_auth_token_shape
 
 pytestmark = pytest.mark.integration
@@ -126,3 +129,40 @@ async def test_me_returns_admin_info(client, auth_headers):
     body = resp.json()
     assert body["username"] == "admin"
     assert body["is_admin"] is True
+
+
+async def test_api_key_owner_is_synced_on_username_rename(client, auth_headers):
+    original = f"admin-rename-{uuid.uuid4().hex[:8]}"
+    renamed = f"{original}-new"
+    password = "pw-12345678"
+
+    create_user = await client.post(
+        "/api/v1/auth/users",
+        headers=auth_headers,
+        json={"username": original, "password": password, "is_admin": True},
+    )
+    assert create_user.status_code == 201, create_user.text
+
+    try:
+        user_headers = {"Authorization": f"Bearer {create_access_token(original)}"}
+
+        create_key = await client.post(
+            "/api/v1/auth/apikeys",
+            headers=user_headers,
+            json={"name": "rename-sync-test"},
+        )
+        assert create_key.status_code == 201, create_key.text
+        api_key = create_key.json()["key"]
+
+        rename = await client.patch(
+            f"/api/v1/auth/users/{original}",
+            headers=auth_headers,
+            json={"username": renamed},
+        )
+        assert rename.status_code == 200, rename.text
+
+        admin_call = await client.get("/api/v1/auth/users", headers={"X-API-Key": api_key})
+        assert admin_call.status_code == 200, admin_call.text
+    finally:
+        await client.delete(f"/api/v1/auth/users/{renamed}", headers=auth_headers)
+        await client.delete(f"/api/v1/auth/users/{original}", headers=auth_headers)

tests/integration/test_icons.py

@@ -7,11 +7,13 @@
 from __future__ import annotations
 
 import io
+import uuid
 import zipfile
 from unittest.mock import patch
 
 import pytest
 import pytest_asyncio
+from obs.api.auth import create_access_token
 
 pytestmark = pytest.mark.integration
 
@@ -159,6 +161,23 @@ async def test_import_zip_skips_non_svg_members(client, auth_headers, icons_tmp)
     assert body["skipped"] >= 1
 
 
+@pytest.mark.asyncio
+async def test_import_zip_is_atomic_on_sanitize_error(client, auth_headers, icons_tmp):
+    malformed_svg = b"<svg><path></svg"
+    zip_bytes = _make_zip(
+        ("ok.svg", _MINIMAL_SVG),
+        ("broken.svg", malformed_svg),
+    )
+    resp = await client.post(
+        "/api/v1/icons/import",
+        headers=auth_headers,
+        files=[("files", ("mixed-broken.zip", zip_bytes, "application/zip"))],
+    )
+    assert resp.status_code == 422
+    assert not (icons_tmp / "ok.svg").exists()
+    assert not (icons_tmp / "broken.svg").exists()
+
+
 @pytest.mark.asyncio
 async def test_import_invalid_zip(client, auth_headers, icons_tmp):
     resp = await client.post(
@@ -169,6 +188,48 @@ async def test_import_invalid_zip(client, auth_headers, icons_tmp):
     assert resp.status_code == 400
 
 
+@pytest.mark.asyncio
+async def test_import_svg_with_admin_api_key(client, auth_headers, icons_tmp):
+    create_key = await client.post(
+        "/api/v1/auth/apikeys",
+        headers=auth_headers,
+        json={"name": "icons-admin-import"},
+    )
+    assert create_key.status_code == 201, create_key.text
+    api_key = create_key.json()["key"]
+
+    resp = await client.post(
+        "/api/v1/icons/import",
+        headers={"X-API-Key": api_key},
+        files=[("files", ("home.svg", _MINIMAL_SVG, "image/svg+xml"))],
+    )
+    assert resp.status_code == 200, resp.text
+    assert (icons_tmp / "home.svg").exists()
+
+
+@pytest.mark.asyncio
+async def test_import_svg_with_non_admin_user(client, auth_headers, icons_tmp):
+    username = f"icons-user-{uuid.uuid4().hex[:8]}"
+    password = "pw-12345678"
+    create_user = await client.post(
+        "/api/v1/auth/users",
+        headers=auth_headers,
+        json={"username": username, "password": password, "is_admin": False},
+    )
+    assert create_user.status_code == 201, create_user.text
+    try:
+        user_headers = {"Authorization": f"Bearer {create_access_token(username)}"}
+        resp = await client.post(
+            "/api/v1/icons/import",
+            headers=user_headers,
+            files=[("files", ("user-icon.svg", _MINIMAL_SVG, "image/svg+xml"))],
+        )
+        assert resp.status_code == 200, resp.text
+        assert (icons_tmp / "user-icon.svg").exists()
+    finally:
+        await client.delete(f"/api/v1/auth/users/{username}", headers=auth_headers)
+
+
 # ---------------------------------------------------------------------------
 # GET /icons/{name} — get single icon
 # ---------------------------------------------------------------------------

tests/unit/test_icons.py

@@ -6,7 +6,7 @@
 
 from __future__ import annotations
 
-from obs.api.v1.icons import _is_svg, _safe_name
+from obs.api.v1.icons import _is_svg, _safe_name, _sanitize_svg
 
 # ---------------------------------------------------------------------------
 # _is_svg
@@ -119,3 +119,46 @@ def test_zip_member_basename(self):
         # _safe_name(member) — dadurch wird der Slash vorher entfernt.
         member = "folder/home.svg"
         assert _safe_name(Path(member).name) == "home"
+
+
+class TestSanitizeSvg:
+    def test_removes_event_handlers(self):
+        payload = b'<svg xmlns="http://www.w3.org/2000/svg" onload="alert(1)"><path d="M1 1"/></svg>'
+        out = _sanitize_svg(payload).decode("utf-8")
+        assert "onload" not in out
+
+    def test_removes_script_and_foreignobject(self):
+        payload = (
+            b'<svg xmlns="http://www.w3.org/2000/svg"><script>alert(1)</script><foreignObject><div>bad</div></foreignObject><path d="M1 1"/></svg>'
+        )
+        out = _sanitize_svg(payload).decode("utf-8")
+        assert "<script" not in out
+        assert "<foreignObject" not in out
+        assert "path" in out
+
+    def test_rejects_invalid_xml(self):
+        import pytest
+        from fastapi import HTTPException
+
+        with pytest.raises(HTTPException):
+            _sanitize_svg(b"<svg><path></svg")
+
+    def test_strips_obfuscated_javascript_href(self):
+        payload = b'<svg xmlns="http://www.w3.org/2000/svg"><a href="java&#10;script:alert(1)"><path d="M1 1"/></a></svg>'
+        out = _sanitize_svg(payload).decode("utf-8")
+        assert "href=" not in out
+
+    def test_rejects_too_deep_svg(self):
+        import pytest
+        from fastapi import HTTPException
+
+        deep = "<svg>" + ("<g>" * 300) + ("</g>" * 300) + "</svg>"
+        with pytest.raises(HTTPException) as exc_info:
+            _sanitize_svg(deep.encode("utf-8"))
+        assert exc_info.value.status_code == 422
+
+    def test_preserves_plain_svg_root_tag(self):
+        payload = b'<svg xmlns="http://www.w3.org/2000/svg"><path d="M1 1"/></svg>'
+        out = _sanitize_svg(payload).decode("utf-8")
+        assert out.startswith("<svg")
+        assert "<ns0:svg" not in out