tests/test_security.py
Rama main · réplica de lectura del repositorio original.
"""Output encoding, input validation and response headers.
The escaping tests use a ticker containing `</script>`. That string is the
whole point: `json.dumps` does not escape it, so embedding chart data with
`json.dumps(...) | safe` let a ticker close the script block early and start a
new one — a stored cross-site scripting hole reachable by anyone with write
access to a shared portfolio. Jinja's `|tojson` escapes `<`, `>` and `&`, which
is why the templates use it.
"""
from __future__ import annotations
import re
import pytest
from sqlalchemy import select
import factories as fac
from app import main as main_module
from app import tenancy
from app.models import Instrument, exchange_problem, ticker_problem
from test_routes import bind_to_only_portfolio, make_login, reading, session_csrf
BREAKOUT = "</script><script>alert(1)</script>"
# The REAL refresh, captured at import: `conftest._no_network` replaces
# `main.refresh_quotes_soon` with a no-op for every test.
_REAL_REFRESH = main_module.refresh_quotes_soon
# --------------------------------------------------------------------------- #
# Response headers
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("header,expected", [
("X-Content-Type-Options", "nosniff"),
("X-Frame-Options", "DENY"),
("Referrer-Policy", "same-origin"),
])
def test_every_response_carries_the_hardening_headers(client, session_factory,
header, expected):
make_login(client, session_factory)
assert client.get("/", headers={"accept": "text/html"}).headers[header] == expected
def _cookie(response, name: str) -> dict:
"""One Set-Cookie header, as {attribute (lower case): value}."""
for header in response.headers.get_list("set-cookie"):
first, *attrs = [part.strip() for part in header.split(";")]
if first.startswith(name + "="):
return {a.split("=", 1)[0].lower(): (a.split("=", 1)[1] if "=" in a else True)
for a in attrs}
raise AssertionError(f"no {name} cookie set")
@pytest.mark.parametrize("secure", [False, True])
def test_the_session_cookie_cannot_be_read_by_a_script(client, session_factory, app_module,
monkeypatch, secure):
"""HttpOnly is what keeps a script that slips past the escaping from
reading the session; nothing checked it. Secure follows the setting."""
from fastapi.testclient import TestClient
from app.auth import hash_password
from test_routes import PASSWORD, pre_auth_csrf
monkeypatch.setattr(app_module.settings.auth, "cookie_secure", secure)
with session_factory() as s:
fac.make_user(s, "user@example.test", password_hash=hash_password(PASSWORD))
s.commit()
# A secure cookie is set only over HTTPS (decisions.md #55); `client` has
# already pointed the app at this test's database.
browser = TestClient(app_module.app, base_url="https://testserver") if secure else client
resp = browser.post("/login", follow_redirects=False, headers={"accept": "text/html"},
data={"email": "user@example.test", "password": PASSWORD,
"_csrf": pre_auth_csrf(browser)})
cookie = _cookie(resp, app_module.settings.auth.cookie_name)
assert cookie.get("httponly") is True
assert cookie.get("samesite", "").lower() == "lax"
assert cookie.get("path") == "/"
assert int(cookie["max-age"]) == app_module.settings.auth.session_ttl_days * 86400
assert (cookie.get("secure") is True) is secure
def test_the_headers_are_on_unauthenticated_responses_too(client):
"""The login page is the one an attacker reaches first."""
assert client.get("/login", headers={"accept": "text/html"}
).headers["X-Frame-Options"] == "DENY"
def test_the_content_security_policy_locks_down_the_dangerous_directives(client,
session_factory):
make_login(client, session_factory)
policy = client.get("/", headers={"accept": "text/html"}).headers[
"Content-Security-Policy"]
# Nothing may frame this app, no injected form may post elsewhere, and no
# injected <base> may re-point every relative URL.
assert "frame-ancestors 'none'" in policy
assert "form-action 'self'" in policy
assert "base-uri 'self'" in policy
assert "default-src 'self'" in policy
def test_the_policy_is_honest_about_inline_scripts(client, session_factory):
"""Recorded, not hidden: the templates still carry inline handlers, so the
policy cannot be the XSS backstop it superficially resembles. If someone
extracts those fragments, this test is the reminder to tighten it."""
make_login(client, session_factory)
policy = client.get("/", headers={"accept": "text/html"}).headers[
"Content-Security-Policy"]
assert "script-src 'self' 'unsafe-inline'" in policy
# --------------------------------------------------------------------------- #
# Ticker validation
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("ticker", ["ACME", "ZULU", "BRK.B", "RDS-A", "X", "ABC123"])
def test_ordinary_tickers_are_accepted(ticker):
assert ticker_problem(ticker) is None
@pytest.mark.parametrize("ticker", [
"", " ",
"A" * 13, # over the 12-character column
"</script>", # the breakout attempt
"AC ME", # whitespace inside
"acme;DROP", # punctuation
".LEADING", # must start alphanumeric
])
def test_malformed_tickers_are_refused(ticker):
assert ticker_problem(ticker) is not None
def test_a_too_long_ticker_is_refused_before_it_reaches_either_backend():
"""SQLite ignores a VARCHAR length and Postgres raises a raw database
error, so the two disagree unless the check happens in the app."""
assert "12 characters" in ticker_problem("A" * 40)
@pytest.mark.parametrize("exchange", ["ASX", "NASDAQ", "NYSE", "CRYPTO"])
def test_ordinary_exchanges_are_accepted(exchange):
assert exchange_problem(exchange) is None
@pytest.mark.parametrize("exchange", ["", "1ASX", "A" * 13, "AS X"])
def test_malformed_exchanges_are_refused(exchange):
assert exchange_problem(exchange) is not None
def test_the_add_instrument_form_refuses_a_malformed_ticker(client, session_factory):
make_login(client, session_factory)
client.post("/holdings/add",
data={"ticker": BREAKOUT, "exchange": "ASX", "asset_class": "etf",
"currency": "AUD", "_csrf": session_csrf(session_factory)},
headers={"accept": "text/html"})
with reading(session_factory) as s:
assert s.scalars(select(Instrument)).all() == []
def test_the_api_refuses_a_malformed_ticker(client, session_factory):
from app import auth as auth_mod
from app.models import ApiKey, Portfolio, User
make_login(client, session_factory)
raw, key_hash, prefix = auth_mod.new_api_key()
with session_factory() as s:
portfolio = s.scalars(select(Portfolio)).first()
owner = s.scalars(select(User)).first()
s.add(ApiKey(portfolios=[portfolio], name="k", key_hash=key_hash,
prefix=prefix, scopes="read,write", created_by=owner.id))
s.commit()
resp = client.post("/api/v1/trades",
json={"ticker": BREAKOUT, "type": "buy", "date": "2026-01-05",
"units": "1", "unit_price": "1.00"},
headers={"Authorization": f"Bearer {raw}"})
assert resp.status_code == 400
# --------------------------------------------------------------------------- #
# Escaping — the data blocks the templates embed
# --------------------------------------------------------------------------- #
def _plant_hostile_instrument(session_factory) -> None:
"""A holding whose *name* carries a script breakout.
The ticker is validated now, so the name is the field a hostile member of a
shared portfolio could still use — and names are rendered into the chart
payloads.
"""
with session_factory() as s:
bind_to_only_portfolio(s)
inst = fac.make_instrument(s, "ACME", name=BREAKOUT)
fac.add_trade(s, inst, "2026-01-05", "buy", 10, "5.00")
fac.add_prices(s, inst, [("2026-08-01", "6.00"), ("2026-08-02", "6.50")])
s.commit()
@pytest.mark.parametrize("path", ["/charts", "/charts/build?edit={chart}", "/holding/ACME"])
def test_a_script_breakout_in_stored_data_is_escaped(client, session_factory, path):
"""The instrument's name is shown on its holding page and a saved chart's
on the two chart pages. Every place each appears is checked: the old
check passed if any encoded bracket was anywhere on the page, and the
chart pages never showed the name at all."""
from app.models import SavedChart
make_login(client, session_factory)
_plant_hostile_instrument(session_factory)
with session_factory() as s:
bind_to_only_portfolio(s)
chart = tenancy.owned(s, SavedChart(name=BREAKOUT, position=0, spec={
"grain": "positions", "x": "ticker", "measures": ["pos_value"], "type": "bar"}))
s.add(chart)
s.commit()
chart_id = chart.id
body = client.get(path.format(chart=chart_id), headers={"accept": "text/html"}).text
# The literal closing tag must never appear — that is what ends the block.
assert "</script><script>alert(1)</script>" not in body
# Wherever the value is rendered, its brackets are encoded: as an entity
# in markup, or as \u003c in a JSON block.
before = [body[max(0, at - 30):at] for at in
(m.start() for m in re.finditer(re.escape("alert(1)"), body))]
assert before, "the premise: the planted value is on the page"
assert all(b.endswith(("<script>", "\\u003cscript\\u003e")) for b in before), before
def test_the_chart_payload_is_still_valid_json(client, session_factory):
"""Escaping must not corrupt the data — the page has to keep working."""
import json
import re
make_login(client, session_factory)
_plant_hostile_instrument(session_factory)
body = client.get("/charts", headers={"accept": "text/html"}).text
block = re.search(r'<script id="chart-data" type="application/json">(.*?)</script>',
body, re.S)
assert block is not None
json.loads(block.group(1)) # parses, breakout and all
# --------------------------------------------------------------------------- #
# Secrets out of URLs
# --------------------------------------------------------------------------- #
def test_a_new_accounts_password_never_appears_in_a_url(client, session_factory):
"""A query string is written to the access log, the browser history and
every proxy in between, so a temporary password must not travel in one."""
make_login(client, session_factory, admin=True)
resp = client.post("/users/add",
data={"name": "New Person", "email": "new@example.test",
"password": "temporary-secret-1", "is_admin": "",
"_csrf": session_csrf(session_factory)},
headers={"accept": "text/html"}, follow_redirects=False)
assert resp.status_code == 200 # rendered, not redirected
assert "temporary-secret-1" in resp.text # …and shown exactly once, in the body
assert "password=" not in str(resp.url)
assert "location" not in resp.headers
def test_a_reset_password_never_appears_in_a_url(client, session_factory):
from app.models import User
make_login(client, session_factory, admin=True)
with session_factory() as s:
other = fac.make_user(s, "other@example.test")
s.commit()
other_id = other.id
resp = client.post(f"/users/{other_id}/reset",
data={"password": "temporary-secret-2",
"_csrf": session_csrf(session_factory)},
headers={"accept": "text/html"}, follow_redirects=False)
assert resp.status_code == 200
assert "temporary-secret-2" in resp.text
assert "location" not in resp.headers
with session_factory() as s:
assert s.scalar(select(User).where(User.id == other_id)).must_change_password
# --------------------------------------------------------------------------- #
# Chart ownership and robustness
# --------------------------------------------------------------------------- #
def test_reordering_charts_survives_rubbish_ids(client, session_factory):
"""The order is posted as JSON from the browser; a malformed id should be
ignored rather than returning a 500."""
make_login(client, session_factory)
client.get("/charts", headers={"accept": "text/html"})
resp = client.post("/charts/order",
json={"order": ["not-a-number", None, 999999]},
headers={"X-CSRF-Token": session_csrf(session_factory)})
assert resp.status_code == 200
# --------------------------------------------------------------------------- #
# Signing in with a password: what a mutation run found unchecked
# --------------------------------------------------------------------------- #
def _seed_failures(session_factory, email: str, count: int = 8) -> None:
"""The configured allowance of failures, from a different address each."""
import datetime as dt
from app.models import LoginAttempt
with session_factory() as s:
for n in range(count):
s.add(LoginAttempt(email=email, ip=f"192.0.2.{n + 1}", success=False,
created_at=dt.datetime.now(dt.timezone.utc)))
s.commit()
def test_a_locked_account_refuses_even_the_right_password(client, session_factory, app_module):
"""`auth.is_locked` has its own tests; nothing signed in while locked, so
the check in the route could go and passwords be guessed without limit."""
from test_routes import PASSWORD, make_login, pre_auth_csrf
email = make_login(client, session_factory)
client.cookies.clear()
_seed_failures(session_factory, email)
resp = client.post("/login", follow_redirects=False, headers={"accept": "text/html"},
data={"email": email, "password": PASSWORD,
"_csrf": pre_auth_csrf(client)})
assert resp.status_code == 200 and "Too many attempts" in resp.text
assert app_module.settings.auth.cookie_name not in resp.cookies
def test_an_email_is_matched_however_it_is_typed(client, session_factory, app_module):
from test_routes import PASSWORD, make_login, pre_auth_csrf
make_login(client, session_factory, email="user@example.test")
client.cookies.clear()
resp = client.post("/login", follow_redirects=False, headers={"accept": "text/html"},
data={"email": " User@Example.TEST ", "password": PASSWORD,
"_csrf": pre_auth_csrf(client)})
assert resp.status_code == 303
assert app_module.settings.auth.cookie_name in resp.cookies
def test_a_password_hashed_to_an_older_standard_is_rehashed_at_sign_in(client, session_factory):
from argon2 import PasswordHasher
from app import auth
from app.models import User
from test_routes import PASSWORD, pre_auth_csrf
old = PasswordHasher(time_cost=1, memory_cost=8, parallelism=1).hash(PASSWORD)
with session_factory() as s:
fac.make_user(s, "user@example.test", password_hash=old)
s.commit()
assert auth.needs_rehash(old)
client.post("/login", follow_redirects=False, headers={"accept": "text/html"},
data={"email": "user@example.test", "password": PASSWORD,
"_csrf": pre_auth_csrf(client)})
with session_factory() as s:
stored = s.query(User).one().password_hash
assert stored != old and not auth.needs_rehash(stored)
assert auth.verify_password(stored, PASSWORD)
def test_a_password_hash_that_is_current_is_left_alone_at_sign_in(client, session_factory):
"""Rehashing is for an older standard; doing it every time would rewrite
the stored hash on every sign-in for nothing."""
from app import auth
from app.models import User
from test_routes import PASSWORD, pre_auth_csrf
current = auth.hash_password(PASSWORD)
with session_factory() as s:
fac.make_user(s, "user@example.test", password_hash=current)
s.commit()
resp = client.post("/login", follow_redirects=False, headers={"accept": "text/html"},
data={"email": "user@example.test", "password": PASSWORD,
"_csrf": pre_auth_csrf(client)})
assert resp.status_code == 303, "the premise: the sign-in went through"
with session_factory() as s:
assert s.query(User).one().password_hash == current
@pytest.mark.parametrize("secure", [False, True])
def test_the_remembered_portfolio_cookie_is_httponly_and_kept_a_year(client, session_factory,
app_module, monkeypatch,
secure):
from fastapi.testclient import TestClient
from app.auth import hash_password
from test_routes import PASSWORD, pre_auth_csrf
monkeypatch.setattr(app_module.settings.auth, "cookie_secure", secure)
with session_factory() as s:
user = fac.make_user(s, "user@example.test", password_hash=hash_password(PASSWORD))
fac.make_portfolio(s, "Mine", owner=user)
s.commit()
browser = TestClient(app_module.app, base_url="https://testserver") if secure else client
resp = browser.post("/login", follow_redirects=False, headers={"accept": "text/html"},
data={"email": "user@example.test", "password": PASSWORD,
"_csrf": pre_auth_csrf(browser)})
cookie = _cookie(resp, app_module.settings.auth.cookie_name + "_portfolio")
assert cookie.get("httponly") is True
assert cookie.get("samesite", "").lower() == "lax"
assert cookie.get("path") == "/"
assert int(cookie["max-age"]) == 365 * 86400
assert (cookie.get("secure") is True) is secure
def test_the_sign_in_forms_csrf_cookie_is_httponly(client, session_factory):
"""A double-submit token: the page carries the same value in a hidden
field, which is what every script reads, so none needs the cookie."""
from app import auth
with session_factory() as s:
fac.make_user(s, "user@example.test") # past the first-run wizard
s.commit()
cookie = _cookie(client.get("/login", headers={"accept": "text/html"}),
auth.PRE_AUTH_CSRF_COOKIE)
assert cookie.get("httponly") is True
def test_a_sign_in_with_no_portfolio_to_open_remembers_none(client, session_factory,
app_module):
"""Nothing to remember is no cookie, not one holding the word "None"."""
from app.auth import hash_password
from test_routes import PASSWORD, pre_auth_csrf
with session_factory() as s:
fac.make_user(s, "user@example.test", password_hash=hash_password(PASSWORD))
s.commit()
resp = client.post("/login", follow_redirects=False, headers={"accept": "text/html"},
data={"email": "user@example.test", "password": PASSWORD,
"_csrf": pre_auth_csrf(client)})
assert resp.status_code == 303, "the premise: the sign-in went through"
remembered = app_module.settings.auth.cookie_name + "_portfolio="
assert not [c for c in resp.headers.get_list("set-cookie") if c.startswith(remembered)]
def test_signing_in_asks_for_live_prices(client, session_factory, app_module, monkeypatch):
"""The one line every way in shares: without it the first page shows
yesterday's close for up to a whole polling interval."""
from test_routes import make_login
asked = []
monkeypatch.setattr(app_module, "refresh_quotes_soon", lambda: asked.append(True))
make_login(client, session_factory)
assert asked == [True]
def test_a_reason_sent_back_to_the_sign_in_page_is_shown(client, session_factory):
"""A provider sign-in happens off-site; when it fails, the sign-in page
is the only place the reason can be said."""
import re
with session_factory() as s:
fac.make_user(s, "user@example.test") # past the first-run wizard
s.commit()
page = client.get("/login", params={"error": "The provider refused the sign-in."},
headers={"accept": "text/html"}).text
assert re.search(r'class="panel error">\s*The provider refused the sign-in\.', page)