28 files

custom_components/taskmate/http_photos.py

@@ -20,12 +20,51 @@ from homeassistant.components.http import HomeAssistantView
 from homeassistant.core import HomeAssistant
 
 from . import photos
+from .const import DOMAIN
 
 _LOGGER = logging.getLogger(__name__)
 
 HTTP_VIEWS_REGISTERED = "photo_http_registered"
 
 
+def _get_coordinator(hass: HomeAssistant):
+    from .coordinator import TaskMateCoordinator
+
+    for value in hass.data.get(DOMAIN, {}).values():
+        if isinstance(value, TaskMateCoordinator):
+            return value
+    return None
+
+
+async def _may_view_photo(hass: HomeAssistant, request: web.Request, filename: str) -> bool:
+    """Whether the requesting user may read this evidence photo.
+
+    Evidence photos can contain images of children, so being merely logged in is
+    not enough: allow an admin, a configured TaskMate parent, or the child whose
+    own completion references the file (via their linked HA user). Signed
+    ``<img>`` requests resolve to the signing user, so per-user client-side
+    signing keeps this bound to the actual viewer.
+    """
+    user = request.get("hass_user")
+    if user is None:
+        return False
+    if getattr(user, "is_admin", False):
+        return True
+    coordinator = _get_coordinator(hass)
+    if coordinator is None:
+        return False
+    if user.id in (coordinator.storage.get_parent_user_ids() or []):
+        return True
+    photo_url = f"{photos.URL_PREFIX}/{filename}"
+    for comp in coordinator.storage.get_completions():
+        if getattr(comp, "photo_url", "") != photo_url:
+            continue
+        child = coordinator.get_child(comp.child_id)
+        if child and getattr(child, "linked_user_id", "") == user.id:
+            return True
+    return False
+
+
 class TaskMatePhotoUploadView(HomeAssistantView):
     """Receive a multipart image upload and store it under the config dir."""
 
@@ -45,9 +84,14 @@ class TaskMatePhotoUploadView(HomeAssistantView):
         except (ValueError, AssertionError):
             return self.json_message("Expected multipart form", HTTPStatus.BAD_REQUEST)
 
-        # Find the "file" part.
+        # Find the "file" part. Bound the scan so a stream of endlessly-named
+        # non-"file" parts can't hold a handler open indefinitely.
         field = await reader.next()
+        parts_scanned = 0
         while field is not None and field.name != "file":
+            parts_scanned += 1
+            if parts_scanned > 16:
+                return self.json_message("Too many form parts", HTTPStatus.BAD_REQUEST)
             field = await reader.next()
         if field is None:
             return self.json_message("No file provided", HTTPStatus.BAD_REQUEST)
@@ -101,6 +145,11 @@ class TaskMatePhotoServeView(HomeAssistantView):
         if not photos.FILENAME_RE.match(filename):
             return web.Response(status=HTTPStatus.NOT_FOUND)
 
+        # Return 404 (not 403) when unauthorized so the endpoint doesn't confirm
+        # a file exists to a caller who may not read it.
+        if not await _may_view_photo(self.hass, request, filename):
+            return web.Response(status=HTTPStatus.NOT_FOUND)
+
         path = photos.photos_path(self.hass) / filename
 
         def _read() -> bytes | None:
@@ -116,7 +165,11 @@ class TaskMatePhotoServeView(HomeAssistantView):
         return web.Response(
             body=data,
             content_type=photos.content_type_for(filename),
-            headers={"Cache-Control": "private, max-age=31536000"},
+            headers={
+                "Cache-Control": "private, max-age=31536000",
+                "X-Content-Type-Options": "nosniff",
+                "Content-Disposition": "inline",
+            },
         )