patx/projectpay
Harden invoice app security
Commit 3157085 · patx · 2026-07-05T23:06:32-04:00
Comments
No comments yet.
Diff
diff --git a/README.md b/README.md
index afaf178..070b9f0 100644
--- a/README.md
+++ b/README.md
@@ -46,8 +46,8 @@ ProjectPay is a small web app for creating project payment pages and collecting
4. Create a `.env` file:
```bash
+ APP_ENV=development
ADMIN_PASSWORD=change-me
- ADMIN_COOKIE_SECRET=replace-with-a-long-random-secret
MONGODB_URI=mongodb://localhost:27017
MONGODB_DB=projectpay
APP_BASE_URL=http://127.0.0.1:8000
@@ -73,21 +73,22 @@ ProjectPay is a small web app for creating project payment pages and collecting
| Variable | Required | Default | Description |
| --- | --- | --- | --- |
-| `ADMIN_PASSWORD` | Recommended | `admin` | Password for the admin dashboard. Change this before deploying. |
-| `ADMIN_COOKIE_SECRET` | Recommended | Falls back to `APP_SECRET`, then `ADMIN_PASSWORD` | Secret used to sign the admin cookie. Use a long random value in production. |
-| `APP_SECRET` | Optional | unset | Fallback cookie secret if `ADMIN_COOKIE_SECRET` is not set. |
-| `APP_BASE_URL` | Recommended in production | Derived from request headers | Public base URL used to generate customer payment links and Stripe return URLs. |
+| `APP_ENV` | Required | unset | Set to `development` locally and `production` for deployed environments. Startup fails if this is missing or unrecognized. |
+| `ADMIN_PASSWORD` | Required outside development | `admin` in development only | Password for the admin dashboard. Production startup fails if this is missing or set to `admin`. |
+| `APP_BASE_URL` | Required outside development | Derived from request headers in development only | Public base URL used to generate customer payment links and Stripe return URLs. |
| `PUBLIC_LOGO_URL` | Optional | unset | Logo image URL shown on public project payment pages. No logo is shown when this is empty. |
| `PUBLIC_LOGO_ALT` | Optional | `ProjectPay Logo` | Alt text for the public payment page logo. |
| `MONGODB_URI` | Optional | `mongodb://localhost:27017` | MongoDB connection string. |
| `MONGODB_DB` | Optional | `invoice_maker` | MongoDB database name. Use `projectpay` or another explicit name for new installs. |
| `MONGODB_SERVER_SELECTION_TIMEOUT_MS` | Optional | `5000` | MongoDB connection timeout in milliseconds. |
+| `LOGIN_RATE_LIMIT_ATTEMPTS` | Optional | `5` | Failed login attempts allowed per client IP during the rate-limit window. Set to `0` to disable. |
+| `LOGIN_RATE_LIMIT_WINDOW_SECONDS` | Optional | `900` | Login rate-limit window in seconds. |
| `STRIPE_SECRET_KEY` | Required for payments | unset | Stripe secret API key. |
| `STRIPE_WEBHOOK_SECRET` | Required for webhook processing | unset | Signing secret for the Stripe webhook endpoint. |
| `STRIPE_PRODUCT_ID` | Required for payments | unset | Stripe Product ID used when creating custom-amount prices. |
| `STRIPE_CURRENCY` | Optional | `usd` | Currency for Stripe Checkout and display formatting. |
| `STRIPE_MINIMUM_AMOUNT_CENTS` | Optional | `100` | Minimum customer payment amount in cents. |
-| `HOST` | Optional | `127.0.0.1` | Host used by `python3 app.py`. |
+| `HOST` | Optional | `127.0.0.1` | Host used by `python3 app.py`. The Heroku `Procfile` binds Uvicorn to `0.0.0.0` separately. |
| `PORT` | Optional | `8000` | Port used by `python3 app.py` and the included `Procfile`. |
## Stripe Setup
@@ -133,8 +134,8 @@ web: uvicorn app:app --host 0.0.0.0 --port $PORT
For production, set at least:
+- `APP_ENV=production`
- `ADMIN_PASSWORD`
-- `ADMIN_COOKIE_SECRET`
- `APP_BASE_URL`
- `MONGODB_URI`
- `MONGODB_DB`
@@ -142,7 +143,24 @@ For production, set at least:
- `STRIPE_WEBHOOK_SECRET`
- `STRIPE_PRODUCT_ID`
-Use HTTPS in production so Stripe webhooks and customer payment redirects work reliably.
+Use HTTPS in production so Stripe webhooks, customer payment redirects, and MicroPie's `Secure` session cookie work reliably. If a local browser does not retain the secure session cookie over plain HTTP, use HTTPS locally or a secure tunnel.
+
+### Heroku
+
+The included `Procfile` already binds Uvicorn to `0.0.0.0` and Heroku's dynamic `$PORT`. For Heroku, set config vars explicitly:
+
+```bash
+heroku config:set APP_ENV=production
+heroku config:set ADMIN_PASSWORD='replace-with-a-strong-password'
+heroku config:set APP_BASE_URL='https://your-app.herokuapp.com'
+heroku config:set MONGODB_URI='mongodb+srv://...'
+heroku config:set MONGODB_DB=projectpay
+heroku config:set STRIPE_SECRET_KEY='sk_live_...'
+heroku config:set STRIPE_WEBHOOK_SECRET='whsec_...'
+heroku config:set STRIPE_PRODUCT_ID='prod_...'
+```
+
+On Heroku dynos, login rate limiting uses the final `X-Forwarded-For` value added by Heroku's router instead of the router's internal connection address.
## Customization
@@ -160,13 +178,21 @@ ProjectPay creates and manages these MongoDB collections:
- `checkout_sessions`
- `webhook_events`
- `counters`
+- `sessions`
+- `login_attempts`
Indexes are created automatically when the app first touches the database.
## Security Notes
+- Set `APP_ENV` explicitly; startup fails if it is missing or misspelled.
+- Do not deploy with `APP_ENV=development`.
- Do not deploy with the default `ADMIN_PASSWORD=admin`.
-- Use a long random `ADMIN_COOKIE_SECRET`.
+- Admin sessions are server-side and stored in MongoDB behind MicroPie's opaque `session_id` cookie.
+- Browser POST forms use CSRF tokens stored in the server-side session.
+- Login attempts are rate-limited by client IP.
+- Project search treats search text as a literal string before sending it to MongoDB regex queries.
+- Stripe webhook storage keeps minimized event summaries instead of full raw event payloads.
- Keep Stripe and MongoDB credentials out of source control.
- The public project URL uses a random share token, but anyone with the link can view the project payment page.
diff --git a/app.py b/app.py
index 05c18ed..d3141fe 100644
--- a/app.py
+++ b/app.py
@@ -2,8 +2,10 @@ import asyncio
import hmac
import json
import os
+import re
import secrets
-from datetime import datetime, timezone
+import uuid
+from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from typing import Any, Dict, List, Optional, Tuple
@@ -14,11 +16,15 @@ from markupsafe import Markup, escape
from pymongo import AsyncMongoClient, ReturnDocument
from pymongo.errors import DuplicateKeyError
-from micropie import App
+from micropie import App, SESSION_TIMEOUT, SessionBackend
PAGE_SIZE = 20
-ADMIN_COOKIE = "invoice_admin"
+CSRF_SESSION_KEY = "csrf_token"
+SESSION_COOKIE = "session_id"
+DEFAULT_HOST = "127.0.0.1"
+DEVELOPMENT_ENVS = {"development", "dev", "local", "test"}
+VALID_APP_ENVS = DEVELOPMENT_ENVS | {"production", "staging"}
SUCCESSFUL_CHECKOUT_EVENTS = {
"checkout.session.completed",
"checkout.session.async_payment_succeeded",
@@ -110,9 +116,59 @@ def as_plain_dict(value: Any) -> Dict[str, Any]:
return {}
+class MongoSessionBackend(SessionBackend):
+ def __init__(self, client: AsyncMongoClient, db_name: str) -> None:
+ self.client = client
+ self.db_name = db_name
+
+ @property
+ def collection(self):
+ return self.client[self.db_name].sessions
+
+ async def load(self, session_id: str) -> Dict[str, Any]:
+ if not session_id:
+ return {}
+ now = now_utc()
+ doc = await self.collection.find_one(
+ {"session_id": session_id, "expires_at": {"$gt": now}},
+ {"data": 1},
+ )
+ if not doc:
+ return {}
+ await self.collection.update_one(
+ {"session_id": session_id},
+ {"$set": {"last_accessed_at": now}},
+ )
+ data = doc.get("data")
+ return dict(data) if isinstance(data, dict) else {}
+
+ async def save(self, session_id: str, data: Dict[str, Any], timeout: int) -> None:
+ if not session_id:
+ return
+ if not data:
+ await self.collection.delete_one({"session_id": session_id})
+ return
+
+ now = now_utc()
+ await self.collection.update_one(
+ {"session_id": session_id},
+ {
+ "$set": {
+ "data": dict(data),
+ "expires_at": now + timedelta(seconds=timeout),
+ "last_accessed_at": now,
+ },
+ "$setOnInsert": {
+ "session_id": session_id,
+ "created_at": now,
+ },
+ },
+ upsert=True,
+ )
+
+
class InvoiceApp(App):
def __init__(self) -> None:
- super().__init__()
self.mongo_uri = os.getenv("MONGODB_URI", "mongodb://localhost:27017")
self.mongo_db_name = os.getenv("MONGODB_DB", "invoice_maker")
timeout_ms = int(os.getenv("MONGODB_SERVER_SELECTION_TIMEOUT_MS", "5000"))
@@ -121,17 +177,29 @@ class InvoiceApp(App):
serverSelectionTimeoutMS=timeout_ms,
connect=False,
)
+ super().__init__(
+ session_backend=MongoSessionBackend(self.client, self.mongo_db_name)
+ )
self._indexes_ready = False
self._index_lock = asyncio.Lock()
- self.admin_password = os.getenv("ADMIN_PASSWORD", "admin")
- self.admin_cookie_secret = (
- os.getenv("ADMIN_COOKIE_SECRET")
- or os.getenv("APP_SECRET")
- or self.admin_password
- or "invoice-maker-dev"
+ self.app_env = os.getenv("APP_ENV", "").strip().lower()
+ self.is_development = self.app_env in DEVELOPMENT_ENVS
+ self.is_heroku = bool(os.getenv("DYNO"))
+ configured_admin_password = os.getenv("ADMIN_PASSWORD")
+ self.admin_password = (
+ configured_admin_password
+ if configured_admin_password and configured_admin_password.strip()
+ else ("admin" if self.is_development else "")
)
self.app_base_url = os.getenv("APP_BASE_URL", "").rstrip("/")
+ self.login_rate_limit_attempts = int(
+ os.getenv("LOGIN_RATE_LIMIT_ATTEMPTS", "5") or "5"
+ )
+ self.login_rate_limit_window_seconds = int(
+ os.getenv("LOGIN_RATE_LIMIT_WINDOW_SECONDS", "900") or "900"
+ )
+ self._validate_config()
self.public_logo_url = os.getenv("PUBLIC_LOGO_URL", "").strip()
self.public_logo_alt = os.getenv("PUBLIC_LOGO_ALT", "ProjectPay Logo").strip()
self.currency = os.getenv("STRIPE_CURRENCY", "usd").upper()
@@ -154,6 +222,22 @@ class InvoiceApp(App):
self.env.filters["star_emphasis"] = format_star_emphasis
self.env.globals["money"] = cents_to_money
+ def _validate_config(self) -> None:
+ if self.app_env not in VALID_APP_ENVS:
+ raise RuntimeError(
+ "APP_ENV must be set to one of: "
+ f"{', '.join(sorted(VALID_APP_ENVS))}"
+ )
+ if self.is_development:
+ return
+ password_for_validation = self.admin_password.strip()
+ if not password_for_validation or password_for_validation == "admin":
+ raise RuntimeError(
+ "ADMIN_PASSWORD must be set to a non-default value outside development"
+ )
+ if not self.app_base_url:
+ raise RuntimeError("APP_BASE_URL must be set outside development")
+
async def _db(self):
await self._ensure_indexes()
return self.client[self.mongo_db_name]
@@ -185,6 +269,11 @@ class InvoiceApp(App):
await db.webhook_events.create_index(
"webhook_id", unique=True, sparse=True
)
+ await db.sessions.create_index("session_id", unique=True)
+ await db.sessions.create_index("expires_at", expireAfterSeconds=0)
+ await db.login_attempts.create_index(
+ "expires_at", expireAfterSeconds=0
+ )
self._indexes_ready = True
async def _shutdown_clients(self) -> None:
@@ -271,10 +360,11 @@ class InvoiceApp(App):
"$set": {
"provider": "stripe",
"event_type": event_type,
- "payload": payload,
+ "payload_summary": self._stripe_event_summary(payload),
"processed": False,
"last_received_at": created_at,
},
+ "$unset": {"payload": ""},
"$inc": {"attempts": 1},
},
upsert=True,
@@ -351,7 +441,8 @@ class InvoiceApp(App):
"currency": currency,
"status": "succeeded",
"event_type": payload.get("type", ""),
- "raw_event": payload,
+ "stripe_event_created": payload.get("created"),
+ "livemode": payload.get("livemode"),
"created_at": now_utc(),
}
if checkout_session_id:
@@ -393,6 +484,55 @@ class InvoiceApp(App):
return {"received": True, "duplicate": True}
return {"received": True, "payment_recorded": True}
+ def _stripe_event_summary(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+ data = as_plain_dict(payload.get("data"))
+ stripe_object = as_plain_dict(data.get("object"))
+ summary: Dict[str, Any] = {
+ "id": payload.get("id"),
+ "type": payload.get("type"),
+ "created": payload.get("created"),
+ "livemode": payload.get("livemode"),
+ "api_version": payload.get("api_version"),
+ }
+
+ request_info = as_plain_dict(payload.get("request"))
+ if request_info:
+ summary["request"] = {
+ key: request_info[key]
+ for key in ("id", "idempotency_key")
+ if request_info.get(key)
+ }
+
+ object_summary = {
+ key: stripe_object[key]
+ for key in (
+ "id",
+ "object",
+ "payment_status",
+ "amount_total",
+ "currency",
+ "payment_intent",
+ "client_reference_id",
+ )
+ if stripe_object.get(key) is not None
+ }
+ metadata = as_plain_dict(stripe_object.get("metadata"))
+ metadata_summary = {
+ key: str(metadata[key])
+ for key in ("project_id", "project_number", "share_token")
+ if metadata.get(key)
+ }
+ if metadata_summary:
+ object_summary["metadata"] = metadata_summary
+ if object_summary:
+ summary["object"] = object_summary
+
+ return {
+ key: value
+ for key, value in summary.items()
+ if value not in (None, "", {})
+ }
+
async def _project_from_stripe_session(
self, session: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
@@ -490,42 +630,129 @@ class InvoiceApp(App):
"balance_cents": balance_cents,
}
- def _signed_admin_cookie(self) -> str:
- signature = hmac.new(
- self.admin_cookie_secret.encode("utf-8"),
- b"admin",
- "sha256",
- ).hexdigest()
- return f"admin.{signature}"
-
- def _cookie_header(self, max_age: int) -> str:
- value = self._signed_admin_cookie() if max_age > 0 else ""
- return (
- f"{ADMIN_COOKIE}={value}; Path=/; Max-Age={max_age}; "
- "SameSite=Lax; HttpOnly"
- )
-
def _is_admin(self) -> bool:
- cookie_header = self.request.headers.get("cookie", "")
- cookies: Dict[str, str] = {}
- for part in cookie_header.split(";"):
- if "=" in part:
- key, value = part.strip().split("=", 1)
- cookies[key] = value
- actual = cookies.get(ADMIN_COOKIE, "")
- return hmac.compare_digest(actual, self._signed_admin_cookie())
+ return bool(self.request.session.get("is_admin"))
def _require_admin(self):
if not self._is_admin():
return self._redirect("/login")
return None
+ def _csrf_token(self) -> str:
+ token = self.request.session.get(CSRF_SESSION_KEY)
+ if not token:
+ token = secrets.token_urlsafe(32)
+ self.request.session[CSRF_SESSION_KEY] = token
+ return str(token)
+
+ def _valid_csrf(self) -> bool:
+ expected = self.request.session.get(CSRF_SESSION_KEY)
+ actual = self.request.form("csrf_token", "")
+ if not isinstance(expected, str) or not isinstance(actual, str) or not expected:
+ return False
+ try:
+ return hmac.compare_digest(
+ expected.encode("utf-8"),
+ actual.encode("utf-8"),
+ )
+ except (TypeError, UnicodeError):
+ return False
+
+ def _csrf_error(self) -> Tuple[int, str]:
+ return 403, "Invalid CSRF token"
+
+ def _session_cookie_header(self, session_id: str, max_age: Optional[int] = None) -> str:
+ parts = [
+ f"{SESSION_COOKIE}={session_id}",
+ "Path=/",
+ "SameSite=Lax",
+ "HttpOnly",
+ "Secure",
+ ]
+ if max_age is not None:
+ parts.insert(2, f"Max-Age={max_age}")
+ return "; ".join(parts)
+
+ async def _start_admin_session(self):
+ session = {
+ "is_admin": True,
+ CSRF_SESSION_KEY: secrets.token_urlsafe(32),
+ }
+ session_id = str(uuid.uuid4())
+ await self.session_backend.save(session_id, session, SESSION_TIMEOUT)
+ self.request.session.clear()
+ return self._redirect("/", [("Set-Cookie", self._session_cookie_header(session_id))])
+
+ def _client_ip(self) -> str:
+ if self.is_heroku:
+ forwarded_for = self.request.headers.get("x-forwarded-for", "")
+ if forwarded_for:
+ return forwarded_for.split(",")[-1].strip()
+
+ client = self.request.scope.get("client") or ("unknown", 0)
+ if isinstance(client, (list, tuple)) and client:
+ return str(client[0])
+ return "unknown"
+
+ def _login_rate_key(self) -> str:
+ return f"login:{self._client_ip()}"
+
+ async def _login_rate_limited(self) -> bool:
+ if self.login_rate_limit_attempts <= 0:
+ return False
+ db = await self._db()
+ doc = await db.login_attempts.find_one(
+ {
+ "_id": self._login_rate_key(),
+ "expires_at": {"$gt": now_utc()},
+ },
+ {"attempts": 1},
+ )
+ return bool(
+ doc and int(doc.get("attempts", 0)) >= self.login_rate_limit_attempts
+ )
+
+ async def _record_failed_login(self) -> None:
+ if self.login_rate_limit_attempts <= 0:
+ return
+ now = now_utc()
+ db = await self._db()
+ await db.login_attempts.delete_many(
+ {"_id": self._login_rate_key(), "expires_at": {"$lte": now}}
+ )
+ await db.login_attempts.update_one(
+ {"_id": self._login_rate_key()},
+ {
+ "$inc": {"attempts": 1},
+ "$set": {
+ "last_failed_at": now,
+ "ip": self._client_ip(),
+ },
+ "$setOnInsert": {
+ "created_at": now,
+ "expires_at": now
+ + timedelta(seconds=self.login_rate_limit_window_seconds),
+ },
+ },
+ upsert=True,
+ )
+
+ async def _clear_failed_logins(self) -> None:
+ db = await self._db()
+ await db.login_attempts.delete_one({"_id": self._login_rate_key()})
+
def _base_url(self) -> str:
if self.app_base_url:
return self.app_base_url
+ if not self.is_development:
+ raise RuntimeError("APP_BASE_URL must be set outside development")
headers = self.request.headers
- proto = headers.get("x-forwarded-proto") or "http"
- host = headers.get("x-forwarded-host") or headers.get("host") or "127.0.0.1:8000"
+ proto = (headers.get("x-forwarded-proto") or "http").split(",", 1)[0].strip()
+ if proto not in {"http", "https"}:
+ proto = "http"
+ host = (
+ headers.get("x-forwarded-host") or headers.get("host") or "127.0.0.1:8000"
+ ).split(",", 1)[0].strip()
return f"{proto}://{host}".rstrip("/")
def _public_project_url(self, project: Dict[str, Any]) -> str:
@@ -574,10 +801,11 @@ class InvoiceApp(App):
query: Dict[str, Any] = {}
q = (q or "").strip()
if q:
+ pattern = re.escape(q)
clauses: List[Dict[str, Any]] = [
- {"project_number": {"$regex": q, "$options": "i"}},
- {"status": {"$regex": q, "$options": "i"}},
- {"customer_name": {"$regex": q, "$options": "i"}},
+ {"project_number": {"$regex": pattern, "$options": "i"}},
+ {"status": {"$regex": pattern, "$options": "i"}},
+ {"customer_name": {"$regex": pattern, "$options": "i"}},
]
oid = object_id(q)
if oid is not None:
@@ -673,6 +901,7 @@ class InvoiceApp(App):
async def _render(self, template: str, **kwargs: Any) -> str:
kwargs.setdefault("admin", self._is_admin())
+ kwargs.setdefault("csrf_token", self._csrf_token())
kwargs.setdefault("currency", self.currency)
kwargs.setdefault("public_logo_url", self.public_logo_url)
kwargs.setdefault("public_logo_alt", self.public_logo_alt)
@@ -707,12 +936,19 @@ class InvoiceApp(App):
async def login(self) -> Any:
if self.request.method == "POST":
+ if not self._valid_csrf():
+ return self._csrf_error()
+ if await self._login_rate_limited():
+ body = await self._render(
+ "login.html",
+ error="Too many failed sign-in attempts. Try again later.",
+ )
+ return 429, body
password = self.request.form("password", "") or ""
if hmac.compare_digest(password, self.admin_password):
- return self._redirect(
- "/",
- [("Set-Cookie", self._cookie_header(30 * 24 * 60 * 60))],
- )
+ await self._clear_failed_logins()
+ return await self._start_admin_session()
+ await self._record_failed_login()
return await self._render("login.html", error="Invalid password.")
if self._is_admin():
@@ -720,7 +956,11 @@ class InvoiceApp(App):
return await self._render("login.html")
def logout(self) -> Any:
- return self._redirect("/login", [("Set-Cookie", self._cookie_header(0))])
+ self.request.session.clear()
+ return self._redirect(
+ "/login",
+ [("Set-Cookie", self._session_cookie_header("", max_age=0))],
+ )
async def new(self) -> Any:
if redirect := self._require_admin():
@@ -739,6 +979,8 @@ class InvoiceApp(App):
return redirect
if self.request.method != "POST":
return self._redirect("/new")
+ if not self._valid_csrf():
+ return self._csrf_error()
project = self._project_from_form()
if project["errors"]:
@@ -764,6 +1006,8 @@ class InvoiceApp(App):
async def edit(self, project_id: str) -> Any:
if redirect := self._require_admin():
return redirect
+ if self.request.method == "POST" and not self._valid_csrf():
+ return self._csrf_error()
oid = object_id(project_id)
db = await self._db()
project = await db.projects.find_one({"_id": oid}) if oid else None
@@ -816,6 +1060,8 @@ class InvoiceApp(App):
async def delete(self, project_id: str) -> Any:
if redirect := self._require_admin():
return redirect
+ if self.request.method == "POST" and not self._valid_csrf():
+ return self._csrf_error()
oid = object_id(project_id)
db = await self._db()
project = await db.projects.find_one({"_id": oid}) if oid else None
@@ -862,6 +1108,8 @@ class InvoiceApp(App):
if self.request.method != "POST":
return self._redirect(f"/p/{share_token}")
+ if not self._valid_csrf():
+ return self._csrf_error()
if summary["balance_cents"] <= 0:
return self._redirect(f"/p/{share_token}")
@@ -902,6 +1150,6 @@ if __name__ == "__main__":
uvicorn.run(
app,
- host=os.getenv("HOST", "127.0.0.1"),
+ host=os.getenv("HOST", DEFAULT_HOST),
port=int(os.getenv("PORT", "8000")),
)
diff --git a/templates/detail.html b/templates/detail.html
index 050f9dc..b2a4869 100644
--- a/templates/detail.html
+++ b/templates/detail.html
@@ -60,6 +60,7 @@
</section>
<form class="full-width-form" method="post" action="/delete/{{ project.id }}" onsubmit="return confirm('Delete {{ project.project_number }}? This cannot be undone.');">
+ <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<button class="danger" type="submit">Delete</button>
</form>
diff --git a/templates/form.html b/templates/form.html
index 75ffcb5..81bde16 100644
--- a/templates/form.html
+++ b/templates/form.html
@@ -21,6 +21,7 @@
{% endif %}
<form method="post" action="{{ '/edit/' ~ project.id if mode == 'edit' else '/create' }}" id="project-form">
+ <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<div class="grid-2">
<div class="field">
<label for="customer_name">Customer Name</label>
diff --git a/templates/login.html b/templates/login.html
index bc909d5..d35e52e 100644
--- a/templates/login.html
+++ b/templates/login.html
@@ -8,6 +8,7 @@
<div class="alert error">{{ error }}</div>
{% endif %}
<form method="post" action="/login">
+ <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<div class="field">
<label for="password">Password</label>
<input id="password" name="password" type="password" autofocus required>
diff --git a/templates/public_project.html b/templates/public_project.html
index 464790f..dfb941e 100644
--- a/templates/public_project.html
+++ b/templates/public_project.html
@@ -36,6 +36,7 @@
{% if project.balance_cents > 0 %}
<form class="sticky-pay" method="post" action="/pay/{{ project.share_token }}">
+ <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<div class="sticky-pay-inner">
<button type="submit">Make a Payment</button>
</div>
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..d892d7a
--- /dev/null
+++ b/tests/__init__.py
@@ -0,0 +1 @@
+# Test package marker for unittest discovery.
diff --git a/tests/test_security_hardening.py b/tests/test_security_hardening.py
new file mode 100644
index 0000000..9bfac84
--- /dev/null
+++ b/tests/test_security_hardening.py
@@ -0,0 +1,206 @@
+import asyncio
+import os
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import micropie
+
+with patch.dict(os.environ, {"APP_ENV": "test"}, clear=False):
+ import app as invoice_app
+
+
+class SecurityHardeningTests(unittest.TestCase):
+ def make_app(self, **env):
+ defaults = {
+ "APP_ENV": "development",
+ "MONGODB_URI": "mongodb://localhost:27017",
+ "MONGODB_DB": "test_projectpay",
+ }
+ defaults.update(env)
+ with patch.dict(os.environ, defaults, clear=True):
+ return invoice_app.InvoiceApp()
+
+ def bind_request(self, app, request):
+ token = micropie.current_request.set(request)
+ self.addCleanup(micropie.current_request.reset, token)
+ return app
+
+ def test_admin_login_starts_fresh_server_side_session(self):
+ class FakeSessionBackend:
+ def __init__(self):
+ self.saved = None
+
+ async def save(self, session_id, data, timeout):
+ self.saved = (session_id, data, timeout)
+
+ app = self.make_app()
+ backend = FakeSessionBackend()
+ app.session_backend = backend
+ request = SimpleNamespace(
+ session={"csrf_token": "old-token"},
+ body_params={},
+ headers={},
+ scope={"client": ("127.0.0.1", 1234)},
+ form=lambda name, default=None: default,
+ )
+ self.bind_request(app, request)
+
+ response = asyncio.run(app._start_admin_session())
+
+ self.assertEqual(response[0], 302)
+ self.assertEqual(request.session, {})
+ self.assertIsNotNone(backend.saved)
+ session_id, data, timeout = backend.saved
+ self.assertTrue(session_id)
+ self.assertTrue(data["is_admin"])
+ self.assertTrue(data["csrf_token"])
+ self.assertGreater(timeout, 0)
+ self.assertIn(f"{invoice_app.SESSION_COOKIE}={session_id}", response[2][1][1])
+
+ def test_logout_expires_session_cookie(self):
+ app = self.make_app()
+ request = SimpleNamespace(
+ session={"is_admin": True},
+ body_params={},
+ headers={},
+ scope={"client": ("127.0.0.1", 1234)},
+ form=lambda name, default=None: default,
+ )
+ self.bind_request(app, request)
+
+ response = app.logout()
+
+ self.assertEqual(request.session, {})
+ self.assertEqual(response[0], 302)
+ self.assertIn(f"{invoice_app.SESSION_COOKIE}=", response[2][1][1])
+ self.assertIn("Max-Age=0", response[2][1][1])
+
+ def test_production_requires_non_default_admin_password(self):
+ with patch.dict(
+ os.environ,
+ {
+ "APP_ENV": "production",
+ "ADMIN_PASSWORD": "admin",
+ "APP_BASE_URL": "https://example.com",
+ },
+ clear=True,
+ ):
+ with self.assertRaisesRegex(RuntimeError, "ADMIN_PASSWORD"):
+ invoice_app.InvoiceApp()
+
+ def test_app_env_must_be_explicit(self):
+ with patch.dict(os.environ, {}, clear=True):
+ with self.assertRaisesRegex(RuntimeError, "APP_ENV"):
+ invoice_app.InvoiceApp()
+
+ def test_production_requires_app_base_url(self):
+ with patch.dict(
+ os.environ,
+ {
+ "APP_ENV": "production",
+ "ADMIN_PASSWORD": "strong-password",
+ "APP_BASE_URL": "",
+ },
+ clear=True,
+ ):
+ with self.assertRaisesRegex(RuntimeError, "APP_BASE_URL"):
+ invoice_app.InvoiceApp()
+
+ def test_csrf_token_is_session_backed_and_validated(self):
+ app = self.make_app()
+ request = SimpleNamespace(
+ session={},
+ body_params={},
+ headers={},
+ scope={"client": ("127.0.0.1", 1234)},
+ form=lambda name, default=None: default,
+ )
+ self.bind_request(app, request)
+
+ csrf_token = app._csrf_token()
+ self.assertTrue(csrf_token)
+ self.assertFalse(app._valid_csrf())
+
+ request.form = lambda name, default=None: {
+ "csrf_token": csrf_token,
+ }.get(name, default)
+ self.assertTrue(app._valid_csrf())
+
+ def test_malformed_csrf_token_is_rejected_without_error(self):
+ app = self.make_app()
+ request = SimpleNamespace(
+ session={invoice_app.CSRF_SESSION_KEY: "ascii-token"},
+ body_params={},
+ headers={},
+ scope={"client": ("127.0.0.1", 1234)},
+ form=lambda name, default=None: {
+ "csrf_token": "snowman-\u2603",
+ }.get(name, default),
+ )
+ self.bind_request(app, request)
+
+ self.assertFalse(app._valid_csrf())
+
+ def test_direct_run_host_defaults_to_loopback(self):
+ self.assertEqual(invoice_app.DEFAULT_HOST, "127.0.0.1")
+
+ def test_development_base_url_uses_request_headers_only_in_development(self):
+ app = self.make_app(APP_BASE_URL="")
+ request = SimpleNamespace(
+ session={},
+ body_params={},
+ headers={"host": "local.test:8000", "x-forwarded-proto": "gopher"},
+ scope={"client": ("127.0.0.1", 1234)},
+ form=lambda name, default=None: default,
+ )
+ self.bind_request(app, request)
+
+ self.assertEqual(app._base_url(), "http://local.test:8000")
+
+ def test_heroku_client_ip_uses_last_forwarded_for_entry(self):
+ app = self.make_app(DYNO="web.1")
+ request = SimpleNamespace(
+ session={},
+ body_params={},
+ headers={"x-forwarded-for": "spoofed, 203.0.113.10"},
+ scope={"client": ("10.0.0.1", 1234)},
+ form=lambda name, default=None: default,
+ )
+ self.bind_request(app, request)
+
+ self.assertEqual(app._client_ip(), "203.0.113.10")
+
+ def test_stripe_event_summary_excludes_full_customer_payload(self):
+ app = self.make_app()
+ summary = app._stripe_event_summary(
+ {
+ "id": "evt_123",
+ "type": "checkout.session.completed",
+ "created": 1234567890,
+ "livemode": False,
+ "data": {
+ "object": {
+ "id": "cs_123",
+ "object": "checkout.session",
+ "payment_status": "paid",
+ "amount_total": 2500,
+ "currency": "usd",
+ "customer_email": "[email protected]",
+ "metadata": {
+ "project_id": "507f1f77bcf86cd799439011",
+ "project_number": "PRJ-2026-0001",
+ "share_token": "share-token",
+ },
+ }
+ },
+ }
+ )
+
+ self.assertEqual(summary["id"], "evt_123")
+ self.assertNotIn("customer_email", str(summary))
+ self.assertEqual(summary["object"]["amount_total"], 2500)
+
+
+if __name__ == "__main__":
+ unittest.main()