fix(web): refuse cross-site state-changing requests (Origin/Referer check) (#674)

The web interface refuses state-changing requests (POST/PUT/PATCH/DELETE)
whose Origin (or, without one, Referer) is not the host they were sent to,
or is null: 403 CROSS_SITE_REQUEST (web_interface/origin_guard.py). Any
website a LAN user visited could otherwise make their browser POST a plain
form to the Pi. /api/v3/system/action also refuses form-encoded and
text/plain bodies (415) unless sent by HTMX. Clients that send no Origin or
Referer (curl, requests, Home Assistant, the MQTT bridge) are unaffected.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-09-30 09:02:01 -04:00
committed by GitHub
co-authored by Claude Opus 5.5
parent c3a7a110c4
commit ba38a83c2c
8 changed files with 565 additions and 10 deletions
+16 -4
View File
@@ -55,10 +55,17 @@ app = Flask(__name__)
app.secret_key = os.urandom(24)
config_manager = ConfigManager()
# No CSRF protection: the UI is meant for the local network, where anyone who
# can forge a request can also send it directly, and neither the HTMX forms
# nor the fetch() calls carry a token. Exposing the UI beyond the LAN needs
# CSRF tokens added to both first.
# Cross-site request forgery: the UI has no login, and being "only on the LAN"
# does not keep other websites out. Any page a LAN user opens can make their
# browser POST to this server -- a plain HTML form is not blocked by CORS -- so
# a hostile site could reboot the Pi, pull code or rewrite the config through
# the user's browser. web_interface/origin_guard.py (registered below) refuses
# POST/PUT/PATCH/DELETE whose Origin (or, failing that, Referer) is not this
# server's own host; requests with neither header (curl, Home Assistant, the
# MQTT bridge) are not from a browser and pass. There are no CSRF tokens:
# neither the HTMX forms nor the fetch() calls carry one. Anyone who can reach
# the port directly can still use the API, so exposing the UI beyond a trusted
# network still needs real authentication.
# Initialize rate limiting (prevent accidental abuse, not security)
try:
@@ -402,6 +409,11 @@ def success_txt():
from web_interface import request_logging
request_logging.init_app(app)
# Refuse state-changing requests sent by another website's page (see the
# cross-site note near the top of this file).
from web_interface import origin_guard
origin_guard.init_app(app)
# Global error handlers
@app.errorhandler(404)
def not_found_error(error):
+16 -5
View File
@@ -383,16 +383,27 @@ def _perform_core_update_locked(stash_local_changes=True):
def execute_system_action():
"""Execute system actions (start/stop/reboot/etc)"""
try:
# HTMX sends data as form data, not JSON
data = request.get_json(silent=True) or {}
if not data:
# Try to get from form data if JSON fails
data = request.get_json(silent=True)
if data is None and not request.is_json:
# Every caller in the interface sends JSON (the Quick Actions
# buttons use HTMX's json-enc). A form-encoded body is what a
# cross-site HTML form can send without a CORS preflight, and
# this route reboots, powers off and pulls code, so it is only
# accepted from HTMX: a cross-site form cannot set HX-Request.
# This backs up the app-wide Origin check (origin_guard.py).
if not request.headers.get('HX-Request'):
return jsonify({
'status': 'error',
'message': ('Send the action as JSON '
'(Content-Type: application/json), '
'e.g. {"action": "restart_display_service"}'),
}), 415
data = {
'action': request.form.get('action'),
'mode': request.form.get('mode')
}
if not data or 'action' not in data:
if not isinstance(data, dict) or not data.get('action'):
return jsonify({'status': 'error', 'message': 'Action required'}), 400
action = data['action']
+191
View File
@@ -0,0 +1,191 @@
"""
Cross-site request guard for the web interface.
The threat: the interface has no login, and "it is only on the LAN" does not
keep other websites out of it. Any page a person on the LAN opens in their
browser can make that browser send a request to ``http://<pi>:5000``. A plain
HTML form POST (``application/x-www-form-urlencoded``, ``multipart/form-data``
or ``text/plain``) is a "simple" request: CORS does not preflight it and does
not stop it from arriving, it only hides the response from the page. So a
hostile or compromised site could reboot the Pi, pull code, install or remove
plugins or rewrite the config, without the user ever seeing the interface.
The defence here needs no tokens and no frontend change. Browsers attach an
``Origin`` header to every cross-site POST (and to same-origin ones in all
current browsers), and it cannot be set or removed by page script. So for any
state-changing method:
* ``Origin`` present -> it must name this server's own host, else 403.
``Origin: null`` (a sandboxed iframe, a ``file://`` page, some cross-site
redirect chains) is never this server, so it is refused too.
* ``Origin`` absent, ``Referer`` present -> the same check on the Referer.
* neither -> allowed. That is curl, Home Assistant, the MQTT bridge and every
other script: not a browser, so not a confused deputy. A browser making a
cross-site request always sends ``Origin``.
"This server's own host" is the ``Host`` header the request arrived with, so
it follows whatever name or address the user typed: ``ledpi.local:5000``,
``192.168.1.40:5000``, or ``192.168.4.1`` in access-point mode (the captive
portal's port 80 -> 5000 redirect keeps the Host the browser sent, and the
setup page's fetches go back to that same host).
The scheme is deliberately not compared, only host and port. The claimed
value's default port comes from its own scheme. A ``Host`` without a port
means "the default port of whatever scheme the browser used", and that scheme
is not always the one Flask sees: a TLS-terminating reverse proxy makes the
browser say ``https://pi.example`` (443) while Flask sees ``http`` (80). So a
portless ``Host`` accepts either default. An attacker cannot use that gap,
because to match they would need to serve a page from this same host on its
standard port. The app does not use ``ProxyFix`` and so does not trust
``X-Forwarded-Host`` or ``X-Forwarded-Proto``: a proxy that rewrites ``Host``
to the upstream address (nginx's default ``proxy_pass`` does) must be
configured to pass the original one, port included
(``proxy_set_header Host $http_host;`` -- nginx's ``$host`` drops the port).
Not covered: DNS rebinding (an attacker's hostname re-pointed at the Pi is
"same origin" to the browser), and anyone who can reach the port directly.
Neither is new; the interface is still meant for a trusted network.
"""
import logging
from urllib.parse import urlsplit, urlunsplit
from flask import Flask, jsonify, request
logger = logging.getLogger('web_interface.origin_guard')
#: Methods that change state and so must come from this interface's own pages.
STATE_CHANGING_METHODS = frozenset({'POST', 'PUT', 'PATCH', 'DELETE'})
_DEFAULT_PORTS = {'http': 80, 'https': 443}
def _authority(netloc: str):
"""``(hostname, port)`` for an authority; port is None when it has none.
Lower-cases the host and drops a trailing dot, so ``Pi.local.`` and
``pi.local`` compare equal. None if the authority is unreadable.
"""
try:
parts = urlsplit(f'//{netloc}')
hostname = parts.hostname
port = parts.port
except ValueError:
# A malformed port or bracketed address.
return None
if not hostname:
return None
return hostname.lower().rstrip('.'), port
def _url_host_port(url: str):
"""``(hostname, port, default_port)`` for an Origin or Referer, or None.
``port`` is the explicit port or, failing that, the URL scheme's default,
which is also returned as ``default_port``.
"""
try:
parts = urlsplit(url.strip())
except ValueError:
return None
scheme = parts.scheme.lower()
if scheme not in _DEFAULT_PORTS or not parts.netloc:
return None
authority = _authority(parts.netloc.rsplit('@', 1)[-1])
if authority is None:
return None
hostname, port = authority
default_port = _DEFAULT_PORTS[scheme]
return hostname, default_port if port is None else port, default_port
def _names_this_server(claimed) -> bool:
"""Whether a claimed ``(hostname, port, default_port)`` is this request's
own ``Host``."""
own = _authority(request.host)
if own is None:
return False
hostname, port = own
claimed_host, claimed_port, claimed_default = claimed
if claimed_host != hostname:
return False
if port is not None:
return claimed_port == port
# A portless Host is the default port of the scheme the browser used.
# Behind a TLS-terminating proxy that is https/443 while Flask sees
# http/80, so accept the default of either scheme.
return claimed_port in (claimed_default,
_DEFAULT_PORTS.get(request.scheme))
def _loggable(value: str) -> str:
"""Just the ``scheme://host[:port]`` of an Origin/Referer, for the log.
A Referer's path and query can carry tokens or other private data, and
only the site matters when reading a refusal.
"""
try:
parts = urlsplit(value.strip())
netloc = parts.netloc.rsplit('@', 1)[-1]
except ValueError:
return '<unreadable>'
if not parts.scheme or not netloc:
return '<unreadable>'
return urlunsplit((parts.scheme, netloc, '', '', ''))
def check_request_origin():
"""None when the request may proceed, else the reason it may not.
The reason is a short phrase for the log and the error message.
"""
if request.method not in STATE_CHANGING_METHODS:
return None
origin = request.headers.get('Origin')
if origin is not None:
header, value = 'Origin', origin
else:
referer = request.headers.get('Referer')
if referer is None:
# Not a browser (curl, Home Assistant, the MQTT bridge, scripts).
return None
header, value = 'Referer', referer
if value.strip().lower() == 'null':
return f'{header} is "null" (sandboxed or file:// page)'
claimed = _url_host_port(value)
if claimed is None:
return f'{header} header is not a valid http(s) URL'
if not _names_this_server(claimed):
# The claimed value is attacker-chosen: the hook logs it, but the
# reason (echoed in the 403 body) never repeats it.
return header + ' names a different host than this interface'
return None
def init_app(app: Flask) -> None:
"""Refuse state-changing requests that another website's page sent."""
@app.before_request
def _refuse_cross_site_requests():
reason = check_request_origin()
if reason is None:
return None
# Only the site each header names, never a Referer's path or query
# (which can carry tokens); %r keeps CR/LF from forging log lines.
origin = request.headers.get('Origin')
referer = request.headers.get('Referer')
logger.warning("Refused cross-site %s %r: %s (Origin=%r, Referer=%r)",
request.method, request.path, reason,
None if origin is None else _loggable(origin),
None if referer is None else _loggable(referer))
return jsonify({
'status': 'error',
'error_code': 'CROSS_SITE_REQUEST',
'message': ('Refused: this request came from another website, not '
'from the LEDMatrix interface. Open the interface '
'directly (the address in your browser bar must be the '
'same one the request goes to) and try again.'),
'details': reason,
}), 403