Sikkerhet: tett path traversal i post-id og herd innlogging/upload
Funn fra sikkerhetsgjennomgang: - HØY: post_id ble ikke validert, så en gjest kunne nå andres innlegg via gjest/../../posts/<slug> (lese utkast, laste opp/slette bilder, publisere eierens utkast). Slug og gjestenavn valideres nå mot samme alfabet de lages med, og stien sjekkes etter oppslag. - Bilder: pikselantall sjekkes før dekoding (maks 40 MP), og DecompressionBombError håndteres. - Innlogging: CSRF-token kreves også på login (login-CSRF), og innloggingsbrems etter 5 feil per brukernavn. - Passordbytte ugyldiggjør alle andre sesjoner (stempel i cookien). - Content-Security-Policy og X-Frame-Options på alle sider; admin har ingen inline script lenger (admin.js). - Gjestenavn og eier-slugs kan ikke lenger kollidere på /navn/. - Healthcheck respekterer SOAPBOX_BASE_PATH; login-redirect beholder base path; run.sh dropper capabilities og setter minnegrense; uv pinnet til 0.11. Rettet også en reell feil: existing_slugs() listet feil katalog (Path / er identitet), så slug-unikhet virket ikke. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JcEy43fNYpwg6K6oTKakWR
This commit is contained in:
parent
d52dea8227
commit
d47558c6c1
14 changed files with 284 additions and 31 deletions
|
|
@ -29,3 +29,5 @@ SOAPBOX_OWNER_PASSWORD=bytt-meg-naa
|
||||||
SOAPBOX_PORT=8080
|
SOAPBOX_PORT=8080
|
||||||
# Podman-volum (navn) eller katalog (absolutt sti) for /data
|
# Podman-volum (navn) eller katalog (absolutt sti) for /data
|
||||||
SOAPBOX_VOLUME=soapbox-data
|
SOAPBOX_VOLUME=soapbox-data
|
||||||
|
# Minnegrense for containeren (bildebehandling er det som bruker mest)
|
||||||
|
SOAPBOX_MEMORY=512m
|
||||||
|
|
|
||||||
|
|
@ -5,7 +5,7 @@
|
||||||
# --build-arg GIT_REVISION="$(git describe --always --dirty)" \
|
# --build-arg GIT_REVISION="$(git describe --always --dirty)" \
|
||||||
# -t soapbox .
|
# -t soapbox .
|
||||||
FROM python:3.13-alpine AS build
|
FROM python:3.13-alpine AS build
|
||||||
COPY --from=ghcr.io/astral-sh/uv:latest /uv /bin/uv
|
COPY --from=ghcr.io/astral-sh/uv:0.11 /uv /bin/uv
|
||||||
WORKDIR /app
|
WORKDIR /app
|
||||||
ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy UV_PYTHON_DOWNLOADS=never
|
ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy UV_PYTHON_DOWNLOADS=never
|
||||||
COPY pyproject.toml uv.lock README.md ./
|
COPY pyproject.toml uv.lock README.md ./
|
||||||
|
|
@ -33,5 +33,5 @@ VOLUME /data
|
||||||
EXPOSE 8080
|
EXPOSE 8080
|
||||||
# waitress avslutter rent på SIGINT; SIGTERM ignoreres når den kjører som PID 1.
|
# waitress avslutter rent på SIGINT; SIGTERM ignoreres når den kjører som PID 1.
|
||||||
STOPSIGNAL SIGINT
|
STOPSIGNAL SIGINT
|
||||||
HEALTHCHECK --interval=30s --timeout=3s CMD curl -fsS http://localhost:8080/ >/dev/null || exit 1
|
HEALTHCHECK --interval=30s --timeout=3s CMD curl -fsS "http://localhost:8080${SOAPBOX_BASE_PATH}/" >/dev/null || exit 1
|
||||||
CMD ["soapbox", "serve"]
|
CMD ["soapbox", "serve"]
|
||||||
|
|
|
||||||
4
run.sh
4
run.sh
|
|
@ -13,12 +13,16 @@ fi
|
||||||
# Verdier som bare gjelder for podman-kjøringen (ikke appen)
|
# Verdier som bare gjelder for podman-kjøringen (ikke appen)
|
||||||
PORT="$(grep -E '^SOAPBOX_PORT=' "$ENV_FILE" | cut -d= -f2- || true)"
|
PORT="$(grep -E '^SOAPBOX_PORT=' "$ENV_FILE" | cut -d= -f2- || true)"
|
||||||
VOLUME="$(grep -E '^SOAPBOX_VOLUME=' "$ENV_FILE" | cut -d= -f2- || true)"
|
VOLUME="$(grep -E '^SOAPBOX_VOLUME=' "$ENV_FILE" | cut -d= -f2- || true)"
|
||||||
|
MEMORY="$(grep -E '^SOAPBOX_MEMORY=' "$ENV_FILE" | cut -d= -f2- || true)"
|
||||||
PORT="${PORT:-8080}"
|
PORT="${PORT:-8080}"
|
||||||
VOLUME="${VOLUME:-soapbox-data}"
|
VOLUME="${VOLUME:-soapbox-data}"
|
||||||
|
MEMORY="${MEMORY:-512m}"
|
||||||
IMAGE="${1:-localhost/soapbox:latest}"
|
IMAGE="${1:-localhost/soapbox:latest}"
|
||||||
|
|
||||||
podman run -d --replace --name soapbox \
|
podman run -d --replace --name soapbox \
|
||||||
--restart unless-stopped \
|
--restart unless-stopped \
|
||||||
|
--cap-drop=all --security-opt=no-new-privileges \
|
||||||
|
--memory "$MEMORY" \
|
||||||
-p "127.0.0.1:${PORT}:8080" \
|
-p "127.0.0.1:${PORT}:8080" \
|
||||||
-v "${VOLUME}:/data:Z" \
|
-v "${VOLUME}:/data:Z" \
|
||||||
--env-file "$ENV_FILE" \
|
--env-file "$ENV_FILE" \
|
||||||
|
|
|
||||||
|
|
@ -15,6 +15,8 @@ import json
|
||||||
import re
|
import re
|
||||||
import secrets
|
import secrets
|
||||||
import sqlite3
|
import sqlite3
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
from functools import wraps
|
from functools import wraps
|
||||||
|
|
||||||
|
|
@ -76,7 +78,40 @@ def _now() -> datetime:
|
||||||
|
|
||||||
def current_user() -> db.User | None:
|
def current_user() -> db.User | None:
|
||||||
uid = session.get("uid")
|
uid = session.get("uid")
|
||||||
return db.get_user(g.db, uid) if uid else None
|
if not uid:
|
||||||
|
return None
|
||||||
|
user = db.get_user(g.db, uid)
|
||||||
|
# Sesjonen bærer et stempel av passord-hashen; bytter passordet, logges
|
||||||
|
# alle eksisterende sesjoner (også stjålne cookies) ut.
|
||||||
|
if user is None or session.get("pw") != db.password_stamp(g.db, uid):
|
||||||
|
return None
|
||||||
|
return user
|
||||||
|
|
||||||
|
|
||||||
|
# Enkel innloggingsbrems i minnet: etter 5 feil må brukernavnet vente
|
||||||
|
# 30 s, deretter dobles ventetiden per feil. Nullstilles ved vellykket login.
|
||||||
|
_login_failures: dict[str, tuple[int, float]] = {}
|
||||||
|
_login_lock = threading.Lock()
|
||||||
|
|
||||||
|
|
||||||
|
def _login_blocked(username: str) -> int:
|
||||||
|
"""Sekunder brukeren må vente, 0 hvis innlogging er tillatt."""
|
||||||
|
with _login_lock:
|
||||||
|
_, until = _login_failures.get(username, (0, 0.0))
|
||||||
|
return max(0, int(until - time.monotonic()))
|
||||||
|
|
||||||
|
|
||||||
|
def _login_failed(username: str) -> None:
|
||||||
|
with _login_lock:
|
||||||
|
count, _ = _login_failures.get(username, (0, 0.0))
|
||||||
|
count += 1
|
||||||
|
wait = 30 * 2 ** (count - 5) if count >= 5 else 0
|
||||||
|
_login_failures[username] = (count, time.monotonic() + wait)
|
||||||
|
|
||||||
|
|
||||||
|
def _login_ok(username: str) -> None:
|
||||||
|
with _login_lock:
|
||||||
|
_login_failures.pop(username, None)
|
||||||
|
|
||||||
|
|
||||||
def csrf_token() -> str:
|
def csrf_token() -> str:
|
||||||
|
|
@ -90,7 +125,7 @@ def login_required(view):
|
||||||
def wrapped(*a, **kw):
|
def wrapped(*a, **kw):
|
||||||
user = current_user()
|
user = current_user()
|
||||||
if user is None:
|
if user is None:
|
||||||
return redirect(url_for("admin.login", next=request.path))
|
return redirect(url_for("admin.login", next=request.script_root + request.path))
|
||||||
g.user = user
|
g.user = user
|
||||||
if request.method == "POST":
|
if request.method == "POST":
|
||||||
token = request.form.get("csrf") or request.headers.get("X-CSRF")
|
token = request.form.get("csrf") or request.headers.get("X-CSRF")
|
||||||
|
|
@ -120,13 +155,23 @@ def _inject():
|
||||||
@bp.route("/login", methods=["GET", "POST"])
|
@bp.route("/login", methods=["GET", "POST"])
|
||||||
def login():
|
def login():
|
||||||
if request.method == "POST":
|
if request.method == "POST":
|
||||||
user = db.authenticate(
|
# CSRF sjekkes også her: ellers kunne en angriper logge offeret inn i
|
||||||
g.db, request.form.get("username", ""), request.form.get("password", "")
|
# angriperens gjestekonto, og alt offeret skriver havner der.
|
||||||
)
|
token = request.form.get("csrf")
|
||||||
|
if not token or token != session.get("csrf"):
|
||||||
|
abort(403)
|
||||||
|
username = request.form.get("username", "")
|
||||||
|
wait = _login_blocked(username)
|
||||||
|
if wait:
|
||||||
|
flash(f"For mange forsøk. Prøv igjen om {wait} sekunder.", "error")
|
||||||
|
return render_template("admin/login.html", next=request.args.get("next", "")), 429
|
||||||
|
user = db.authenticate(g.db, username, request.form.get("password", ""))
|
||||||
if user:
|
if user:
|
||||||
|
_login_ok(username)
|
||||||
session.clear()
|
session.clear()
|
||||||
session.permanent = True
|
session.permanent = True
|
||||||
session["uid"] = user.id
|
session["uid"] = user.id
|
||||||
|
session["pw"] = db.password_stamp(g.db, user.id)
|
||||||
csrf_token()
|
csrf_token()
|
||||||
nxt = request.form.get("next") or url_for("admin.index")
|
nxt = request.form.get("next") or url_for("admin.index")
|
||||||
# Kun stier på dette nettstedet: "//evil.com" og "/\\evil.com"
|
# Kun stier på dette nettstedet: "//evil.com" og "/\\evil.com"
|
||||||
|
|
@ -134,6 +179,7 @@ def login():
|
||||||
if not nxt.startswith("/") or nxt.startswith("//") or "\\" in nxt:
|
if not nxt.startswith("/") or nxt.startswith("//") or "\\" in nxt:
|
||||||
nxt = url_for("admin.index")
|
nxt = url_for("admin.index")
|
||||||
return redirect(nxt)
|
return redirect(nxt)
|
||||||
|
_login_failed(username)
|
||||||
flash("Feil brukernavn eller passord.", "error")
|
flash("Feil brukernavn eller passord.", "error")
|
||||||
return render_template("admin/login.html", next=request.args.get("next", ""))
|
return render_template("admin/login.html", next=request.args.get("next", ""))
|
||||||
|
|
||||||
|
|
@ -149,12 +195,20 @@ def logout():
|
||||||
|
|
||||||
|
|
||||||
def _load_or_404(post_id: str) -> Post:
|
def _load_or_404(post_id: str) -> Post:
|
||||||
guest, slug = split_id(post_id)
|
cfg = _cfg()
|
||||||
post = get_post(_cfg(), guest, slug)
|
try:
|
||||||
|
guest, slug = split_id(post_id)
|
||||||
|
except ValueError:
|
||||||
|
abort(404)
|
||||||
|
post = get_post(cfg, guest, slug)
|
||||||
if post is None:
|
if post is None:
|
||||||
abort(404)
|
abort(404)
|
||||||
if not g.user.is_owner and guest != g.user.username:
|
if not g.user.is_owner and guest != g.user.username:
|
||||||
abort(403)
|
abort(403)
|
||||||
|
# Forsvar i dybden: stien skal ligge nøyaktig der id-en sier.
|
||||||
|
expected = post_dir(cfg, guest, slug).resolve()
|
||||||
|
if post.path.resolve().parent != expected:
|
||||||
|
abort(404)
|
||||||
return post
|
return post
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -217,6 +271,9 @@ def edit(post_id: str):
|
||||||
wanted = slugify(title)
|
wanted = slugify(title)
|
||||||
if wanted != post.slug:
|
if wanted != post.slug:
|
||||||
taken = (existing_slugs(cfg, post.guest) - {post.slug}) | RESERVED
|
taken = (existing_slugs(cfg, post.guest) - {post.slug}) | RESERVED
|
||||||
|
if post.guest is None:
|
||||||
|
# Eierens /slug/ må ikke kollidere med en gjests /brukernavn/
|
||||||
|
taken |= {u.username for u in db.list_users(g.db)}
|
||||||
new_slug = unique_slug(wanted, taken)
|
new_slug = unique_slug(wanted, taken)
|
||||||
post_dir(cfg, post.guest, post.slug).rename(post_dir(cfg, post.guest, new_slug))
|
post_dir(cfg, post.guest, post.slug).rename(post_dir(cfg, post.guest, new_slug))
|
||||||
post.slug = new_slug
|
post.slug = new_slug
|
||||||
|
|
@ -339,6 +396,8 @@ def users():
|
||||||
password = request.form.get("password", "")
|
password = request.form.get("password", "")
|
||||||
if not USERNAME_RE.match(username) or username in RESERVED:
|
if not USERNAME_RE.match(username) or username in RESERVED:
|
||||||
flash("Ugyldig brukernavn (a–z, 0–9, bindestrek, 2–31 tegn).", "error")
|
flash("Ugyldig brukernavn (a–z, 0–9, bindestrek, 2–31 tegn).", "error")
|
||||||
|
elif username in existing_slugs(_cfg(), None):
|
||||||
|
flash("Brukernavnet kolliderer med et innlegg på /" + username + "/.", "error")
|
||||||
elif len(password) < 8:
|
elif len(password) < 8:
|
||||||
flash("Passordet må være minst 8 tegn.", "error")
|
flash("Passordet må være minst 8 tegn.", "error")
|
||||||
else:
|
else:
|
||||||
|
|
@ -378,6 +437,7 @@ def password():
|
||||||
flash("Passordet må være minst 8 tegn.", "error")
|
flash("Passordet må være minst 8 tegn.", "error")
|
||||||
else:
|
else:
|
||||||
db.set_password(g.db, g.user.id, request.form["password"])
|
db.set_password(g.db, g.user.id, request.form["password"])
|
||||||
|
session["pw"] = db.password_stamp(g.db, g.user.id)
|
||||||
flash("Passordet er endret.", "ok")
|
flash("Passordet er endret.", "ok")
|
||||||
return redirect(url_for("admin.index"))
|
return redirect(url_for("admin.index"))
|
||||||
return render_template("admin/password.html")
|
return render_template("admin/password.html")
|
||||||
|
|
|
||||||
|
|
@ -23,6 +23,16 @@ from .admin import bp as admin_bp
|
||||||
from .build import build_site
|
from .build import build_site
|
||||||
from .config import Config, load_config
|
from .config import Config, load_config
|
||||||
|
|
||||||
|
# Admin har ingen inline script/style; alt ligger i egne filer.
|
||||||
|
ADMIN_CSP = (
|
||||||
|
"default-src 'self'; img-src 'self' data: blob:; frame-ancestors 'none'; form-action 'self'"
|
||||||
|
)
|
||||||
|
# Den offentlige siden tillater inline CSS så egne temaer kan bruke style-attributter.
|
||||||
|
PUBLIC_CSP = (
|
||||||
|
"default-src 'self'; img-src 'self' data:; style-src 'self' 'unsafe-inline'; "
|
||||||
|
"script-src 'self'; frame-ancestors 'none'"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def create_app(cfg: Config | None = None) -> Flask:
|
def create_app(cfg: Config | None = None) -> Flask:
|
||||||
cfg = cfg or load_config()
|
cfg = cfg or load_config()
|
||||||
|
|
@ -59,9 +69,12 @@ def create_app(cfg: Config | None = None) -> Flask:
|
||||||
def _headers(resp):
|
def _headers(resp):
|
||||||
resp.headers.setdefault("X-Content-Type-Options", "nosniff")
|
resp.headers.setdefault("X-Content-Type-Options", "nosniff")
|
||||||
resp.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin")
|
resp.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin")
|
||||||
|
resp.headers.setdefault("X-Frame-Options", "DENY")
|
||||||
if request.path.startswith("/admin"):
|
if request.path.startswith("/admin"):
|
||||||
resp.headers["Cache-Control"] = "no-store"
|
resp.headers["Cache-Control"] = "no-store"
|
||||||
resp.headers["X-Frame-Options"] = "DENY"
|
resp.headers.setdefault("Content-Security-Policy", ADMIN_CSP)
|
||||||
|
else:
|
||||||
|
resp.headers.setdefault("Content-Security-Policy", PUBLIC_CSP)
|
||||||
return resp
|
return resp
|
||||||
|
|
||||||
app.register_blueprint(admin_bp, url_prefix="/admin")
|
app.register_blueprint(admin_bp, url_prefix="/admin")
|
||||||
|
|
@ -87,7 +100,7 @@ def _serve_public(cfg: Config, path: str):
|
||||||
if not root.exists():
|
if not root.exists():
|
||||||
abort(404)
|
abort(404)
|
||||||
target = (root / path).resolve() if path else root
|
target = (root / path).resolve() if path else root
|
||||||
if not str(target).startswith(str(root.resolve())):
|
if not target.is_relative_to(root.resolve()):
|
||||||
abort(404)
|
abort(404)
|
||||||
if target.is_dir():
|
if target.is_dir():
|
||||||
# Katalog-URL-er må ha avsluttende skråstrek, ellers brekker relative
|
# Katalog-URL-er må ha avsluttende skråstrek, ellers brekker relative
|
||||||
|
|
|
||||||
|
|
@ -18,7 +18,11 @@ def main(argv: list[str] | None = None) -> int:
|
||||||
serve = sub.add_parser("serve", help="Start webserveren")
|
serve = sub.add_parser("serve", help="Start webserveren")
|
||||||
serve.add_argument("--host", default="0.0.0.0")
|
serve.add_argument("--host", default="0.0.0.0")
|
||||||
serve.add_argument("--port", type=int, default=8080)
|
serve.add_argument("--port", type=int, default=8080)
|
||||||
serve.add_argument("--debug", action="store_true", help="Flask dev-server med reload")
|
serve.add_argument(
|
||||||
|
"--debug",
|
||||||
|
action="store_true",
|
||||||
|
help="Flask dev-server med reload og debugger. KUN lokalt – debuggeren gir kodekjøring.",
|
||||||
|
)
|
||||||
|
|
||||||
sub.add_parser("build", help="Generer statiske sider på nytt")
|
sub.add_parser("build", help="Generer statiske sider på nytt")
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -139,7 +139,7 @@ def get_post(cfg: Config, guest: str | None, slug: str) -> Post | None:
|
||||||
|
|
||||||
|
|
||||||
def existing_slugs(cfg: Config, guest: str | None) -> set[str]:
|
def existing_slugs(cfg: Config, guest: str | None) -> set[str]:
|
||||||
base = post_dir(cfg, guest, "").parent
|
base = post_dir(cfg, guest, "x").parent # Path / "" er identitet, derfor "x"
|
||||||
return {p.name for p in base.iterdir() if p.is_dir()} if base.exists() else set()
|
return {p.name for p in base.iterdir() if p.is_dir()} if base.exists() else set()
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -167,7 +167,20 @@ def delete_post(cfg: Config, post: Post) -> None:
|
||||||
shutil.rmtree(d)
|
shutil.rmtree(d)
|
||||||
|
|
||||||
|
|
||||||
|
SLUG_RE = re.compile(r"^[\w-]+$") # samme alfabet som slugify() produserer
|
||||||
|
GUEST_RE = re.compile(r"^[a-z0-9][a-z0-9-]{1,30}$")
|
||||||
|
|
||||||
|
|
||||||
def split_id(post_id: str) -> tuple[str | None, str]:
|
def split_id(post_id: str) -> tuple[str | None, str]:
|
||||||
"""``"_owner/slug"`` → (None, slug); ``"gjest/slug"`` → ("gjest", slug)."""
|
"""``"_owner/slug"`` → (None, slug); ``"gjest/slug"`` → ("gjest", slug).
|
||||||
|
|
||||||
|
Begge delene valideres mot alfabetet de kan ha blitt laget med. Uten dette
|
||||||
|
ville ``gjest/../../posts/hemmelig`` truffet en annen brukers mappe, siden
|
||||||
|
delene brukes direkte som katalognavn.
|
||||||
|
"""
|
||||||
guest, _, slug = post_id.partition("/")
|
guest, _, slug = post_id.partition("/")
|
||||||
|
if not SLUG_RE.match(slug) or "." in slug:
|
||||||
|
raise ValueError("ugyldig slug")
|
||||||
|
if guest != OWNER and not GUEST_RE.match(guest):
|
||||||
|
raise ValueError("ugyldig gjest")
|
||||||
return (None if guest == OWNER else guest), slug
|
return (None if guest == OWNER else guest), slug
|
||||||
|
|
|
||||||
|
|
@ -124,6 +124,13 @@ def authenticate(conn: sqlite3.Connection, username: str, password: str) -> User
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def password_stamp(conn: sqlite3.Connection, user_id: int) -> str | None:
|
||||||
|
"""Kort, ikke-hemmelig fingeravtrykk av passord-hashen, lagret i sesjonen
|
||||||
|
slik at et passordbytte ugyldiggjør alle eksisterende sesjoner."""
|
||||||
|
row = conn.execute("SELECT password_hash FROM users WHERE id=?", (user_id,)).fetchone()
|
||||||
|
return row["password_hash"][-16:] if row else None
|
||||||
|
|
||||||
|
|
||||||
def set_password(conn: sqlite3.Connection, user_id: int, password: str) -> None:
|
def set_password(conn: sqlite3.Connection, user_id: int, password: str) -> None:
|
||||||
conn.execute("UPDATE users SET password_hash=? WHERE id=?", (hash_password(password), user_id))
|
conn.execute("UPDATE users SET password_hash=? WHERE id=?", (hash_password(password), user_id))
|
||||||
conn.commit()
|
conn.commit()
|
||||||
|
|
|
||||||
|
|
@ -16,6 +16,8 @@ from PIL import Image, ImageOps, UnidentifiedImageError
|
||||||
from .slug import slugify
|
from .slug import slugify
|
||||||
|
|
||||||
ALLOWED = {"JPEG": ".jpg", "PNG": ".png", "GIF": ".gif", "WEBP": ".webp"}
|
ALLOWED = {"JPEG": ".jpg", "PNG": ".png", "GIF": ".gif", "WEBP": ".webp"}
|
||||||
|
MAX_PIXELS = 40_000_000 # ~7000×5700; mobilkameraer ligger på 12–50 MP nedskalert
|
||||||
|
Image.MAX_IMAGE_PIXELS = MAX_PIXELS
|
||||||
|
|
||||||
|
|
||||||
class ImageError(ValueError):
|
class ImageError(ValueError):
|
||||||
|
|
@ -36,7 +38,13 @@ def process_image(data: bytes, max_px: int) -> tuple[bytes, str]:
|
||||||
"""Returnerer (bytes, filendelse) for et nedskalert bilde uten EXIF."""
|
"""Returnerer (bytes, filendelse) for et nedskalert bilde uten EXIF."""
|
||||||
try:
|
try:
|
||||||
img = Image.open(io.BytesIO(data))
|
img = Image.open(io.BytesIO(data))
|
||||||
|
# Sjekk oppgitt størrelse FØR pikslene dekodes: en liten PNG kan
|
||||||
|
# deklarere 30000×30000 og spise gigabytes med minne.
|
||||||
|
if img.width * img.height > MAX_PIXELS:
|
||||||
|
raise ImageError("Bildet er for stort (maks 40 megapiksler)")
|
||||||
img.load()
|
img.load()
|
||||||
|
except Image.DecompressionBombError as e:
|
||||||
|
raise ImageError("Bildet er for stort") from e
|
||||||
except (UnidentifiedImageError, OSError) as e:
|
except (UnidentifiedImageError, OSError) as e:
|
||||||
raise ImageError("Filen er ikke et gyldig bilde") from e
|
raise ImageError("Filen er ikke et gyldig bilde") from e
|
||||||
fmt = img.format
|
fmt = img.format
|
||||||
|
|
|
||||||
7
soapbox/static/admin/admin.js
Normal file
7
soapbox/static/admin/admin.js
Normal file
|
|
@ -0,0 +1,7 @@
|
||||||
|
/* Felles admin-JS. Bekreftelsesdialoger via data-confirm i stedet for inline
|
||||||
|
* onsubmit, slik at tekst fra innhold (titler, brukernavn) aldri havner i en
|
||||||
|
* JS-kontekst – og slik at CSP kan forby inline script. */
|
||||||
|
document.addEventListener("submit", function (ev) {
|
||||||
|
var msg = ev.target.dataset && ev.target.dataset.confirm;
|
||||||
|
if (msg && !confirm(msg)) ev.preventDefault();
|
||||||
|
});
|
||||||
|
|
@ -33,13 +33,6 @@
|
||||||
{% endwith %}
|
{% endwith %}
|
||||||
{% block content %}{% endblock %}
|
{% block content %}{% endblock %}
|
||||||
</main>
|
</main>
|
||||||
<script>
|
<script src="{{ url_for('admin.static', filename='admin.js') }}"></script>
|
||||||
// Bekreftelsesdialoger via data-attributt i stedet for inline onsubmit, slik at
|
|
||||||
// tekst fra innhold (titler, brukernavn) aldri havner i en JS-kontekst.
|
|
||||||
document.addEventListener("submit", function (ev) {
|
|
||||||
var msg = ev.target.dataset && ev.target.dataset.confirm;
|
|
||||||
if (msg && !confirm(msg)) ev.preventDefault();
|
|
||||||
});
|
|
||||||
</script>
|
|
||||||
</body>
|
</body>
|
||||||
</html>
|
</html>
|
||||||
|
|
|
||||||
|
|
@ -32,7 +32,11 @@ def client(app):
|
||||||
|
|
||||||
|
|
||||||
def login(client, username="eier", password="hemmelig123", prefix=""):
|
def login(client, username="eier", password="hemmelig123", prefix=""):
|
||||||
return client.post(f"{prefix}/admin/login", data={"username": username, "password": password})
|
client.get(f"{prefix}/admin/login") # gir sesjonen et CSRF-token
|
||||||
|
return client.post(
|
||||||
|
f"{prefix}/admin/login",
|
||||||
|
data={"username": username, "password": password, "csrf": csrf(client, prefix)},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def csrf(client, prefix=""):
|
def csrf(client, prefix=""):
|
||||||
|
|
|
||||||
|
|
@ -114,15 +114,14 @@ def test_guest_flow_and_isolation(client, cfg):
|
||||||
|
|
||||||
|
|
||||||
def test_login_next_no_open_redirect(client):
|
def test_login_next_no_open_redirect(client):
|
||||||
|
def attempt(nxt):
|
||||||
|
client.get("/admin/login") # ferskt CSRF-token (roteres ved innlogging)
|
||||||
|
creds = {"username": "eier", "password": "hemmelig123", "csrf": csrf(client)}
|
||||||
|
return client.post("/admin/login", data={**creds, "next": nxt})
|
||||||
|
|
||||||
for bad in ("//evil.com", "/\\evil.com", "https://evil.com"):
|
for bad in ("//evil.com", "/\\evil.com", "https://evil.com"):
|
||||||
r = client.post(
|
assert attempt(bad).headers["Location"] == "/admin/", bad
|
||||||
"/admin/login", data={"username": "eier", "password": "hemmelig123", "next": bad}
|
assert attempt("/admin/users").headers["Location"] == "/admin/users"
|
||||||
)
|
|
||||||
assert r.headers["Location"] == "/admin/", bad
|
|
||||||
r = client.post(
|
|
||||||
"/admin/login", data={"username": "eier", "password": "hemmelig123", "next": "/admin/users"}
|
|
||||||
)
|
|
||||||
assert r.headers["Location"] == "/admin/users"
|
|
||||||
|
|
||||||
|
|
||||||
def test_csrf_enforced(client):
|
def test_csrf_enforced(client):
|
||||||
|
|
@ -233,3 +232,18 @@ def test_rebuild_on_startup(cfg):
|
||||||
(cfg.public_dir / "index.html").write_text("GAMMEL")
|
(cfg.public_dir / "index.html").write_text("GAMMEL")
|
||||||
create_app(cfg)
|
create_app(cfg)
|
||||||
assert "GAMMEL" not in (cfg.public_dir / "index.html").read_text()
|
assert "GAMMEL" not in (cfg.public_dir / "index.html").read_text()
|
||||||
|
|
||||||
|
|
||||||
|
def test_slug_uniqueness(client, cfg):
|
||||||
|
login(client)
|
||||||
|
token = csrf(client)
|
||||||
|
ids = []
|
||||||
|
for _ in range(2):
|
||||||
|
r = client.post("/admin/new", data={"csrf": token})
|
||||||
|
pid = r.headers["Location"].split("/admin/edit/")[1]
|
||||||
|
r = client.post(
|
||||||
|
f"/admin/edit/{pid}",
|
||||||
|
data={"csrf": token, "title": "Samme", "body": "x", "action": "publish"},
|
||||||
|
)
|
||||||
|
ids.append(r.headers["Location"].split("/admin/edit/")[1])
|
||||||
|
assert ids == ["_owner/samme", "_owner/samme-2"]
|
||||||
|
|
|
||||||
124
tests/test_security.py
Normal file
124
tests/test_security.py
Normal file
|
|
@ -0,0 +1,124 @@
|
||||||
|
"""Regresjonstester for funn fra sikkerhetsgjennomgangen."""
|
||||||
|
|
||||||
|
import io
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from PIL import Image
|
||||||
|
|
||||||
|
from soapbox import db
|
||||||
|
from soapbox.images import ImageError, process_image
|
||||||
|
from tests.conftest import csrf, login
|
||||||
|
|
||||||
|
|
||||||
|
def _owner_draft(client, cfg):
|
||||||
|
login(client)
|
||||||
|
r = client.post("/admin/new", data={"csrf": csrf(client)})
|
||||||
|
pid = r.headers["Location"].split("/admin/edit/")[1]
|
||||||
|
client.post(
|
||||||
|
f"/admin/edit/{pid}",
|
||||||
|
data={"csrf": csrf(client), "title": "Hemmelig", "body": "TOPPHEMMELIG", "action": "save"},
|
||||||
|
)
|
||||||
|
(cfg.content_dir / "posts/hemmelig/priv.png").write_bytes(b"x")
|
||||||
|
client.post("/admin/logout", data={"csrf": csrf(client)})
|
||||||
|
return "_owner/hemmelig"
|
||||||
|
|
||||||
|
|
||||||
|
def test_guest_cannot_traverse_to_owner_post(client, cfg):
|
||||||
|
_owner_draft(client, cfg)
|
||||||
|
login(client, "gjest1")
|
||||||
|
client.post("/admin/new", data={"csrf": csrf(client)}) # så guests/gjest1/ finnes
|
||||||
|
token = csrf(client)
|
||||||
|
for bad in (
|
||||||
|
"gjest1/../../posts/hemmelig",
|
||||||
|
"gjest1/%2e%2e/%2e%2e/posts/hemmelig",
|
||||||
|
"_owner/hemmelig",
|
||||||
|
"../posts/hemmelig",
|
||||||
|
):
|
||||||
|
assert client.get(f"/admin/edit/{bad}").status_code in (403, 404), bad
|
||||||
|
assert client.get(f"/admin/history/{bad}").status_code in (403, 404), bad
|
||||||
|
assert client.get(f"/admin/media/{bad}/priv.png").status_code in (403, 404), bad
|
||||||
|
r = client.post(
|
||||||
|
f"/admin/upload/{bad}",
|
||||||
|
data={"file": (io.BytesIO(b"x"), "e.png")},
|
||||||
|
headers={"X-CSRF": token},
|
||||||
|
content_type="multipart/form-data",
|
||||||
|
)
|
||||||
|
assert r.status_code in (403, 404), bad
|
||||||
|
assert "TOPPHEMMELIG" not in client.get("/admin/edit/gjest1/../../posts/hemmelig").text
|
||||||
|
assert not (cfg.content_dir / "posts/hemmelig/e.png").exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_login_requires_csrf(client):
|
||||||
|
r = client.post("/admin/login", data={"username": "eier", "password": "hemmelig123"})
|
||||||
|
assert r.status_code == 403
|
||||||
|
|
||||||
|
|
||||||
|
def test_login_backoff(client):
|
||||||
|
client.get("/admin/login")
|
||||||
|
token = csrf(client)
|
||||||
|
for _ in range(5):
|
||||||
|
client.post("/admin/login", data={"username": "eier", "password": "feil", "csrf": token})
|
||||||
|
r = client.post(
|
||||||
|
"/admin/login", data={"username": "eier", "password": "hemmelig123", "csrf": token}
|
||||||
|
)
|
||||||
|
assert r.status_code == 429
|
||||||
|
from soapbox import admin
|
||||||
|
|
||||||
|
admin._login_failures.clear()
|
||||||
|
|
||||||
|
|
||||||
|
def test_password_change_invalidates_other_sessions(app, client, cfg):
|
||||||
|
login(client)
|
||||||
|
other = app.test_client()
|
||||||
|
login(other)
|
||||||
|
assert other.get("/admin/").status_code == 200
|
||||||
|
conn = db.connect(cfg.db_path)
|
||||||
|
db.set_password(conn, db.get_user_by_name(conn, "eier").id, "nyttpassord1")
|
||||||
|
conn.close()
|
||||||
|
assert other.get("/admin/").status_code == 302 # logget ut
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.filterwarnings("ignore::PIL.Image.DecompressionBombWarning")
|
||||||
|
def test_decompression_bomb_rejected():
|
||||||
|
# Liten PNG som deklarerer 8000x8000 = 64 MP > grensen
|
||||||
|
buf = io.BytesIO()
|
||||||
|
Image.new("1", (8000, 8000)).save(buf, "PNG")
|
||||||
|
try:
|
||||||
|
process_image(buf.getvalue(), 2000)
|
||||||
|
except ImageError as e:
|
||||||
|
assert "for stort" in str(e)
|
||||||
|
else:
|
||||||
|
raise AssertionError("skulle vært avvist")
|
||||||
|
|
||||||
|
|
||||||
|
def test_security_headers(client):
|
||||||
|
r = client.get("/")
|
||||||
|
assert "frame-ancestors 'none'" in r.headers["Content-Security-Policy"]
|
||||||
|
assert r.headers["X-Frame-Options"] == "DENY"
|
||||||
|
login(client)
|
||||||
|
assert "unsafe-inline" not in client.get("/admin/").headers["Content-Security-Policy"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_guest_username_cannot_shadow_owner_post(client, cfg):
|
||||||
|
login(client)
|
||||||
|
token = csrf(client)
|
||||||
|
r = client.post("/admin/new", data={"csrf": token})
|
||||||
|
pid = r.headers["Location"].split("/admin/edit/")[1]
|
||||||
|
client.post(
|
||||||
|
f"/admin/edit/{pid}", data={"csrf": token, "title": "foo", "body": "x", "action": "publish"}
|
||||||
|
)
|
||||||
|
client.post(
|
||||||
|
"/admin/users",
|
||||||
|
data={"csrf": token, "action": "create", "username": "foo", "password": "passord123"},
|
||||||
|
)
|
||||||
|
conn = db.connect(cfg.db_path)
|
||||||
|
assert db.get_user_by_name(conn, "foo") is None
|
||||||
|
conn.close()
|
||||||
|
# og motsatt: eierens innlegg kan ikke få slug lik et gjestenavn
|
||||||
|
r = client.post("/admin/new", data={"csrf": token})
|
||||||
|
pid = r.headers["Location"].split("/admin/edit/")[1]
|
||||||
|
r = client.post(
|
||||||
|
f"/admin/edit/{pid}",
|
||||||
|
data={"csrf": token, "title": "gjest1", "body": "x", "action": "publish"},
|
||||||
|
)
|
||||||
|
assert r.headers["Location"].endswith("/admin/edit/_owner/gjest1-2")
|
||||||
Loading…
Add table
Add a link
Reference in a new issue