S1: shared value helpers (parse_dt/coerce_bool) + auto-Secure session cookie
M9 hardening/DRY pass — section S1, commit 1 (the shared-toolkit foundation): - Add src/thoughtsync/common.py with parse_dt() and coerce_bool(): one home for the ISO-date and truthy-flag coercions that were duplicated across modules. notes.py adopts them and deletes _parse_iso_dt, _iso_to_dt and _truthy (rule 22 — old copies removed; callers, incl. tests, updated). - Security: the session cookie is now marked Secure automatically on any request that arrived over HTTPS (directly or via a proxy's X-Forwarded-Proto), via a SecureCookieSessionInterface override. Hardens HTTPS deployments without breaking plain-HTTP LAN installs — no config. Behavior-preserving refactor + one security hardening. The backend serialization layer, the json_error sweep, and the notes.py split follow as their own commits. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FRgehjoz7Yv8LkUfADxACm
This commit is contained in:
+21
-1
@@ -5,7 +5,8 @@ import os
|
||||
import secrets
|
||||
from datetime import timedelta
|
||||
|
||||
from quart import Quart, jsonify, send_from_directory
|
||||
from quart import Quart, has_request_context, jsonify, request, send_from_directory
|
||||
from quart.sessions import SecureCookieSessionInterface
|
||||
|
||||
from . import __version__
|
||||
from .auth import bp as auth_bp
|
||||
@@ -26,6 +27,23 @@ STATIC_DIR = os.path.join(os.path.dirname(__file__), "static")
|
||||
mimetypes.add_type("application/manifest+json", ".webmanifest")
|
||||
|
||||
|
||||
class _AutoSecureSessionInterface(SecureCookieSessionInterface):
|
||||
"""Mark the session cookie `Secure` whenever the request arrived over HTTPS —
|
||||
directly, or via a TLS-terminating reverse proxy that sets X-Forwarded-Proto.
|
||||
|
||||
Auto-detecting per request (rather than a fixed SESSION_COOKIE_SECURE flag)
|
||||
hardens the cookie on HTTPS deployments without breaking a plain-HTTP install on
|
||||
a trusted LAN, where a hard-forced Secure flag would stop the browser from ever
|
||||
sending the cookie back — i.e. silently break login. No configuration required.
|
||||
"""
|
||||
|
||||
def get_cookie_secure(self, app: Quart) -> bool:
|
||||
if not has_request_context():
|
||||
return False
|
||||
forwarded = request.headers.get("X-Forwarded-Proto", "").split(",")[0].strip().lower()
|
||||
return forwarded == "https" or request.is_secure
|
||||
|
||||
|
||||
def create_app() -> Quart:
|
||||
# static_folder=None: the SPA catch-all below owns static serving.
|
||||
app = Quart(__name__, static_folder=None)
|
||||
@@ -35,6 +53,8 @@ def create_app() -> Quart:
|
||||
app.config["APP_VERSION"] = os.environ.get("APP_VERSION", __version__)
|
||||
app.config["SESSION_COOKIE_HTTPONLY"] = True
|
||||
app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
|
||||
# Auto-mark the session cookie Secure on HTTPS requests (see the interface above).
|
||||
app.session_interface = _AutoSecureSessionInterface()
|
||||
app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(days=30)
|
||||
# Hard request-body ceiling (any-file attachments, import zips, sync push). The
|
||||
# per-file attachment limit is the DB-backed `max_attachment_mb` setting, enforced
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
# Small, dependency-free value coercions shared across the blueprints. Kept in one
|
||||
# place so the "parse an ISO date" / "is this flag truthy" logic has a single
|
||||
# definition instead of a near-identical copy per module.
|
||||
|
||||
|
||||
def parse_dt(raw: object) -> datetime | None:
|
||||
"""Parse an ISO-8601 timestamp (accepting a trailing 'Z' for UTC).
|
||||
|
||||
Returns None for anything that isn't a non-empty string or doesn't parse, so
|
||||
callers can treat "absent", "blank", and "malformed" uniformly (a route that
|
||||
wants a 400 on malformed input checks for None itself).
|
||||
"""
|
||||
if not isinstance(raw, str) or not raw:
|
||||
return None
|
||||
try:
|
||||
return datetime.fromisoformat(raw.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def coerce_bool(raw: object) -> bool:
|
||||
"""Truthy for the common flag spellings ('true'/'1'/'yes'/'on', or a real bool).
|
||||
|
||||
Used for query-string booleans (?has_reminder=true) and DB-backed bool settings,
|
||||
which arrive as strings.
|
||||
"""
|
||||
if isinstance(raw, bool):
|
||||
return raw
|
||||
if isinstance(raw, str):
|
||||
return raw.strip().lower() in ("true", "1", "yes", "on")
|
||||
return False
|
||||
+15
-33
@@ -16,6 +16,7 @@ from sqlalchemy import case, delete, func, literal_column, select
|
||||
|
||||
from .acl import visible_to_user
|
||||
from .auth import login_required
|
||||
from .common import coerce_bool, parse_dt
|
||||
from .config import Config
|
||||
from .db import session_scope
|
||||
from .settings import get_setting
|
||||
@@ -320,12 +321,6 @@ async def _rename_inbound_links(db, renamed: Note, old_title: str, new_title: st
|
||||
await _rewrite_links(db, source)
|
||||
|
||||
|
||||
def _parse_iso_dt(raw: str) -> datetime:
|
||||
"""Parse an ISO-8601 timestamp (accepting a trailing 'Z' for UTC), raising
|
||||
ValueError on anything unparseable — used to validate date-range query params."""
|
||||
return datetime.fromisoformat(raw.replace("Z", "+00:00"))
|
||||
|
||||
|
||||
REMINDER_RECURRENCES = {"daily", "weekly", "monthly", "yearly"}
|
||||
|
||||
|
||||
@@ -369,10 +364,6 @@ def next_occurrence(remind_at: datetime, recurrence: str, after: datetime) -> da
|
||||
return nxt
|
||||
|
||||
|
||||
def _truthy(raw: str | None) -> bool:
|
||||
return raw in ("true", "1", "yes", "on")
|
||||
|
||||
|
||||
@bp.get("")
|
||||
@login_required
|
||||
async def list_notes():
|
||||
@@ -384,8 +375,8 @@ async def list_notes():
|
||||
label_params = request.args.getlist("label")
|
||||
color = request.args.get("color")
|
||||
kind = request.args.get("kind")
|
||||
has_reminder = _truthy(request.args.get("has_reminder"))
|
||||
has_attachment = _truthy(request.args.get("has_attachment"))
|
||||
has_reminder = coerce_bool(request.args.get("has_reminder"))
|
||||
has_attachment = coerce_bool(request.args.get("has_attachment"))
|
||||
query_text = (request.args.get("q") or "").strip()
|
||||
# Optional creation-date range — the "browse by when" / Timeline lens. Both bounds
|
||||
# are ISO-8601 instants forming a HALF-OPEN interval [created_after, created_before),
|
||||
@@ -417,15 +408,15 @@ async def list_notes():
|
||||
if has_attachment:
|
||||
stmt = stmt.where(Note.id.in_(select(NoteAttachment.note_id)))
|
||||
if after_param:
|
||||
try:
|
||||
stmt = stmt.where(Note.created_at >= _parse_iso_dt(after_param))
|
||||
except ValueError:
|
||||
after_dt = parse_dt(after_param)
|
||||
if after_dt is None:
|
||||
return jsonify({"error": "invalid created_after"}), 400
|
||||
stmt = stmt.where(Note.created_at >= after_dt)
|
||||
if before_param:
|
||||
try:
|
||||
stmt = stmt.where(Note.created_at < _parse_iso_dt(before_param))
|
||||
except ValueError:
|
||||
before_dt = parse_dt(before_param)
|
||||
if before_dt is None:
|
||||
return jsonify({"error": "invalid created_before"}), 400
|
||||
stmt = stmt.where(Note.created_at < before_dt)
|
||||
if query_text:
|
||||
# Full-text match over title+body (generated tsvector, migration 0005),
|
||||
# ranked — so the facet bar's text box searches, not just filters.
|
||||
@@ -665,15 +656,6 @@ def _usec_to_dt(usec: object) -> datetime | None:
|
||||
return None
|
||||
|
||||
|
||||
def _iso_to_dt(raw: object) -> datetime | None:
|
||||
if not isinstance(raw, str) or not raw:
|
||||
return None
|
||||
try:
|
||||
return _parse_iso_dt(raw)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _native_spec(n: dict) -> dict:
|
||||
"""Normalize one note from a ThoughtSync export's notes.json into the common
|
||||
import spec consumed by _create_imported_note."""
|
||||
@@ -685,10 +667,10 @@ def _native_spec(n: dict) -> dict:
|
||||
"pinned": bool(n.get("pinned")),
|
||||
"archived": bool(n.get("archived")),
|
||||
"trashed": False, # export only includes live notes
|
||||
"remind_at": _iso_to_dt(n.get("remind_at")),
|
||||
"remind_at": parse_dt(n.get("remind_at")),
|
||||
"recurrence": normalize_recurrence(n.get("recurrence")),
|
||||
"created_at": _iso_to_dt(n.get("created_at")),
|
||||
"updated_at": _iso_to_dt(n.get("updated_at")),
|
||||
"created_at": parse_dt(n.get("created_at")),
|
||||
"updated_at": parse_dt(n.get("updated_at")),
|
||||
"labels": [s for s in (n.get("labels") or []) if isinstance(s, str)],
|
||||
"items": [
|
||||
{"text": it.get("text"), "checked": bool(it.get("checked"))}
|
||||
@@ -1133,10 +1115,10 @@ async def update_note(note_id: str):
|
||||
note.remind_at = None
|
||||
note.recurrence = None # no reminder → recurrence is moot
|
||||
else:
|
||||
try:
|
||||
note.remind_at = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
remind_dt = parse_dt(raw)
|
||||
if remind_dt is None:
|
||||
return jsonify({"error": "invalid remind_at"}), 400
|
||||
note.remind_at = remind_dt
|
||||
if "recurrence" in data:
|
||||
note.recurrence = normalize_recurrence(data["recurrence"])
|
||||
# Recompute the display name (explicit title, else first body line) whenever
|
||||
|
||||
+16
-14
@@ -3,15 +3,14 @@ from datetime import datetime, timezone
|
||||
import pytest
|
||||
|
||||
from thoughtsync.app import create_app
|
||||
from thoughtsync.common import coerce_bool, parse_dt
|
||||
from thoughtsync.models.note import NOTE_COLORS, Note
|
||||
from thoughtsync.notes import (
|
||||
_attachment_ext,
|
||||
_escape_like,
|
||||
_header_filename,
|
||||
_truthy,
|
||||
_keep_spec,
|
||||
_native_spec,
|
||||
_parse_iso_dt,
|
||||
_safe_filename,
|
||||
_slugify,
|
||||
_usec_to_dt,
|
||||
@@ -170,18 +169,19 @@ def test_escape_like():
|
||||
assert _escape_like("plain") == "plain"
|
||||
|
||||
|
||||
def test_parse_iso_dt():
|
||||
def test_parse_dt():
|
||||
# A full ISO instant round-trips (used to validate the Timeline date range).
|
||||
d = _parse_iso_dt("2026-07-19T12:30:00+00:00")
|
||||
d = parse_dt("2026-07-19T12:30:00+00:00")
|
||||
assert (d.year, d.month, d.day, d.hour, d.minute) == (2026, 7, 19, 12, 30)
|
||||
assert d.tzinfo is not None
|
||||
# a trailing Z is accepted as UTC
|
||||
assert _parse_iso_dt("2026-07-19T00:00:00Z").tzinfo is not None
|
||||
assert parse_dt("2026-07-19T00:00:00Z").tzinfo is not None
|
||||
# a plain calendar date parses to midnight
|
||||
assert _parse_iso_dt("2026-07-19").hour == 0
|
||||
# garbage raises (the endpoint turns this into a 400)
|
||||
with pytest.raises(ValueError):
|
||||
_parse_iso_dt("not-a-date")
|
||||
assert parse_dt("2026-07-19").hour == 0
|
||||
# garbage / non-strings return None (the endpoint turns this into a 400)
|
||||
assert parse_dt("not-a-date") is None
|
||||
assert parse_dt("") is None
|
||||
assert parse_dt(None) is None
|
||||
|
||||
|
||||
async def test_titles_requires_auth(app):
|
||||
@@ -264,11 +264,13 @@ def test_header_filename():
|
||||
assert _header_filename("") == "file"
|
||||
|
||||
|
||||
def test_truthy():
|
||||
assert _truthy("true") and _truthy("1") and _truthy("yes") and _truthy("on")
|
||||
assert not _truthy("false")
|
||||
assert not _truthy(None)
|
||||
assert not _truthy("")
|
||||
def test_coerce_bool():
|
||||
assert coerce_bool("true") and coerce_bool("1") and coerce_bool("yes") and coerce_bool("on")
|
||||
assert coerce_bool(True)
|
||||
assert not coerce_bool("false")
|
||||
assert not coerce_bool(None)
|
||||
assert not coerce_bool("")
|
||||
assert not coerce_bool(False)
|
||||
|
||||
|
||||
def test_normalize_recurrence():
|
||||
|
||||
Reference in New Issue
Block a user