mirror of
https://github.com/rstrouse/ESPSomfy-RTS.git
synced 2026-10-31 13:52:15 +01:00
fix: close the token bypass below a route, review fixes for 4b
The routes matched any path below them, and a web asset name there passed the token check: GET /backup/a.js returned the backup under Secured. Every route is now an exact match, and a web asset sits in the root only. - e2e: those paths answer 401 or 404; the restore test now fails when the chunks skip the token check (checked with the check removed). - WEB_CORS lets the socket take Origin: null instead of skipping the Origin check. - ADR 0006, GLOSSARY and README: the Origin check in either mode, /ws as a public route, the Host header behind a reverse proxy. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
ecb0260f61
commit
0ee52d6d9f
10 changed files with 105 additions and 51 deletions
|
|
@ -4,13 +4,14 @@ mode (docs/adr/0006-two-security-modes-eleven-http-routes.md).
|
|||
Like test_auth, it tests the mode it finds: pass --password or --pin for a
|
||||
Secured device.
|
||||
"""
|
||||
import json
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
import pytest
|
||||
|
||||
from proto_client import ProtoClient, heap
|
||||
from proto_client import ProtoClient
|
||||
|
||||
# Every route that needs the token under Secured, sent without a body or
|
||||
# parameters, so a route that let one through would still change nothing.
|
||||
|
|
@ -28,23 +29,25 @@ SHADE_FILES = ["/shades.cfg", "/shades.tmp"]
|
|||
|
||||
|
||||
def http(host, path, method="GET", body=None, headers=None):
|
||||
"""The status and headers of a request, refusals included."""
|
||||
"""The status, headers and body of a request, refusals included."""
|
||||
req = urllib.request.Request(f"http://{host}{path}", data=body, method=method, headers=headers or {})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=20) as r:
|
||||
r.read()
|
||||
return r.status, r.headers
|
||||
return r.status, r.headers, r.read()
|
||||
except urllib.error.HTTPError as e:
|
||||
e.read()
|
||||
return e.code, e.headers
|
||||
return e.code, e.headers, e.read()
|
||||
|
||||
|
||||
def upload(host, path, headers=None):
|
||||
"""Posts a small file that isn't a backup or an image."""
|
||||
def upload(host, path, headers=None, with_file=True):
|
||||
"""Posts a small file that isn't a backup or an image, or a form without a file."""
|
||||
boundary = "e2e-upload"
|
||||
body = (
|
||||
f"--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"e2e.bin\"\r\n"
|
||||
"Content-Type: application/octet-stream\r\n\r\n").encode() + b"\0" * 256 + f"\r\n--{boundary}--\r\n".encode()
|
||||
if with_file:
|
||||
part = (
|
||||
f"--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"e2e.bin\"\r\n"
|
||||
"Content-Type: application/octet-stream\r\n\r\n").encode() + b"\0" * 256
|
||||
else:
|
||||
part = f"--{boundary}\r\nContent-Disposition: form-data; name=\"e2e\"\r\n\r\n1".encode()
|
||||
body = part + f"\r\n--{boundary}--\r\n".encode()
|
||||
return http(host, path, "POST", body, {"Content-Type": f"multipart/form-data; boundary={boundary}", **(headers or {})})
|
||||
|
||||
|
||||
|
|
@ -53,7 +56,7 @@ def test_secured_refuses_a_route_without_a_token(host, secured, method, path):
|
|||
if not secured:
|
||||
pytest.skip("the device's security is None")
|
||||
body = b"{}" if method == "PUT" else None
|
||||
status, _ = http(host, path, method, body, {"Content-Type": "application/json"} if body else None)
|
||||
status, _, _ = http(host, path, method, body, {"Content-Type": "application/json"} if body else None)
|
||||
assert status == 401
|
||||
|
||||
|
||||
|
|
@ -65,49 +68,80 @@ def test_secured_refuses_a_route_with_a_wrong_token(host, secured, method, path)
|
|||
headers = {"apikey": "0" * 64}
|
||||
if body:
|
||||
headers["Content-Type"] = "application/json"
|
||||
status, _ = http(host, path, method, body, headers)
|
||||
status, _, _ = http(host, path, method, body, headers)
|
||||
assert status == 401
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method,path", TOKEN_ROUTES + [
|
||||
("POST", "/restore"), ("POST", "/updateFirmware"), ("POST", "/updateApplication")])
|
||||
def test_secured_refuses_a_web_asset_name_below_a_route(host, secured, method, path):
|
||||
"""/backup/a.js once reached /backup and passed the token check as a web asset.
|
||||
|
||||
The uploads are sent without a body: one that got through would reboot
|
||||
the device.
|
||||
"""
|
||||
if not secured:
|
||||
pytest.skip("the device's security is None")
|
||||
body = b"{}" if method == "PUT" else None
|
||||
status, _, _ = http(host, path + "/a.js", method, body, {"Content-Type": "application/json"} if body else None)
|
||||
assert status in (401, 404)
|
||||
|
||||
|
||||
def test_secured_accepts_the_token_on_a_route(host, secured, token):
|
||||
if not secured:
|
||||
pytest.skip("the device's security is None")
|
||||
status, _ = http(host, "/backup", headers={"apikey": token})
|
||||
status, _, _ = http(host, "/backup", headers={"apikey": token})
|
||||
assert status == 200
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", ["/restore", "/updateFirmware"])
|
||||
def test_secured_drops_an_upload_without_a_token(host, secured, token, path):
|
||||
"""The token check runs once the body is in, so the uploads check it on their first chunk.
|
||||
def test_secured_refuses_an_upload_without_a_token(host, secured, path):
|
||||
"""/updateApplication isn't sent: one that got through would write over the file system."""
|
||||
if not secured:
|
||||
pytest.skip("the device's security is None")
|
||||
status, _, _ = upload(host, path)
|
||||
assert status == 401
|
||||
|
||||
An /updateFirmware that got through would stop the radio and reboot the
|
||||
device once the connection closed. /updateApplication isn't sent: one
|
||||
that got through would write over the file system.
|
||||
|
||||
def restore_without_a_file(host, token):
|
||||
status, _, body = upload(host, "/restore", {"apikey": token}, with_file=False)
|
||||
return status, json.loads(body or b"{}").get("desc", "")
|
||||
|
||||
|
||||
def test_secured_drops_the_chunks_of_an_upload_without_a_token(host, secured, token):
|
||||
"""The token check runs once the body is in, so /restore's chunks check the token themselves.
|
||||
|
||||
A chunk that got through would mark the upload complete. /restore with
|
||||
the token and no file would then find the refused file and call it
|
||||
invalid, instead of saying no upload completed. Neither restores anything.
|
||||
"""
|
||||
if not secured:
|
||||
pytest.skip("the device's security is None")
|
||||
before = heap(host, token)
|
||||
status, _ = upload(host, path)
|
||||
status, desc = restore_without_a_file(host, token)
|
||||
if status != 400 or "didn't complete" not in desc:
|
||||
pytest.skip("an earlier upload is still marked complete: reboot the device")
|
||||
status, _, _ = upload(host, "/restore")
|
||||
assert status == 401
|
||||
after = heap(host, token)
|
||||
assert after["uptime"] >= before["uptime"], "the device rebooted"
|
||||
status, desc = restore_without_a_file(host, token)
|
||||
assert status == 400
|
||||
assert "didn't complete" in desc, desc
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", PUBLIC_ROUTES)
|
||||
def test_public_routes_need_no_token(host, path):
|
||||
status, _ = http(host, path)
|
||||
status, _, _ = http(host, path)
|
||||
assert status == 200
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", SHADE_FILES)
|
||||
def test_the_shade_files_are_not_served(host, path):
|
||||
"""Their rolling codes leave the device only through /backup."""
|
||||
status, _ = http(host, path)
|
||||
status, _, _ = http(host, path)
|
||||
assert status in (401, 404)
|
||||
|
||||
|
||||
def test_replies_carry_no_cors_headers(host):
|
||||
_, headers = http(host, "/loginContext", headers={"Origin": "http://example.com"})
|
||||
_, headers, _ = http(host, "/loginContext", headers={"Origin": "http://example.com"})
|
||||
assert headers.get("Access-Control-Allow-Origin") is None
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue