feat(l4d2-web): add login page and safe redirects
This commit is contained in:
parent
942dada807
commit
0aca36506f
8 changed files with 170 additions and 39 deletions
|
|
@ -1,9 +1,9 @@
|
||||||
import os
|
import os
|
||||||
import secrets
|
import secrets
|
||||||
|
|
||||||
from flask import Flask, Response, jsonify, request, session
|
from flask import Flask, Response, jsonify, redirect, request, session
|
||||||
|
|
||||||
from l4d2web.auth import load_current_user
|
from l4d2web.auth import current_user, load_current_user
|
||||||
from l4d2web.cli import register_cli
|
from l4d2web.cli import register_cli
|
||||||
from l4d2web.config import DEFAULT_CONFIG
|
from l4d2web.config import DEFAULT_CONFIG
|
||||||
from l4d2web.db import init_db
|
from l4d2web.db import init_db
|
||||||
|
|
@ -24,7 +24,7 @@ def create_app(test_config: dict[str, object] | None = None) -> Flask:
|
||||||
app.config.update(
|
app.config.update(
|
||||||
SESSION_COOKIE_HTTPONLY=True,
|
SESSION_COOKIE_HTTPONLY=True,
|
||||||
SESSION_COOKIE_SAMESITE="Lax",
|
SESSION_COOKIE_SAMESITE="Lax",
|
||||||
CSRF_EXEMPT_PATHS={"/login", "/signup", "/health"},
|
CSRF_EXEMPT_PATHS={"/login", "/health"},
|
||||||
)
|
)
|
||||||
if test_config is not None:
|
if test_config is not None:
|
||||||
app.config.update(test_config)
|
app.config.update(test_config)
|
||||||
|
|
@ -40,6 +40,9 @@ def create_app(test_config: dict[str, object] | None = None) -> Flask:
|
||||||
if request.method not in {"POST", "PUT", "PATCH", "DELETE"}:
|
if request.method not in {"POST", "PUT", "PATCH", "DELETE"}:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
if request.endpoint is None:
|
||||||
|
return None
|
||||||
|
|
||||||
if request.path in app.config["CSRF_EXEMPT_PATHS"]:
|
if request.path in app.config["CSRF_EXEMPT_PATHS"]:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
@ -67,6 +70,8 @@ def create_app(test_config: dict[str, object] | None = None) -> Flask:
|
||||||
|
|
||||||
@app.get("/")
|
@app.get("/")
|
||||||
def root():
|
def root():
|
||||||
return jsonify({"status": "ok"})
|
if current_user() is None:
|
||||||
|
return redirect("/login")
|
||||||
|
return redirect("/dashboard")
|
||||||
|
|
||||||
return app
|
return app
|
||||||
|
|
|
||||||
|
|
@ -1,7 +1,8 @@
|
||||||
from functools import wraps
|
from functools import wraps
|
||||||
from typing import Callable, TypeVar
|
from typing import Callable, TypeVar
|
||||||
|
from urllib.parse import quote
|
||||||
|
|
||||||
from flask import abort, g, redirect, session
|
from flask import abort, g, redirect, request, session
|
||||||
from sqlalchemy import select
|
from sqlalchemy import select
|
||||||
from werkzeug.security import check_password_hash, generate_password_hash
|
from werkzeug.security import check_password_hash, generate_password_hash
|
||||||
|
|
||||||
|
|
@ -41,11 +42,30 @@ def logout_user() -> None:
|
||||||
session.pop("user_id", None)
|
session.pop("user_id", None)
|
||||||
|
|
||||||
|
|
||||||
|
def is_safe_next(target: str | None) -> bool:
|
||||||
|
if not target:
|
||||||
|
return False
|
||||||
|
if not target.startswith("/"):
|
||||||
|
return False
|
||||||
|
if target.startswith("//"):
|
||||||
|
return False
|
||||||
|
if "://" in target:
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def login_redirect_for_current_request():
|
||||||
|
target = request.full_path.rstrip("?")
|
||||||
|
if is_safe_next(target):
|
||||||
|
return redirect(f"/login?next={quote(target, safe='/')}")
|
||||||
|
return redirect("/login")
|
||||||
|
|
||||||
|
|
||||||
def require_login(func: F) -> F:
|
def require_login(func: F) -> F:
|
||||||
@wraps(func)
|
@wraps(func)
|
||||||
def wrapper(*args, **kwargs):
|
def wrapper(*args, **kwargs):
|
||||||
if current_user() is None:
|
if current_user() is None:
|
||||||
return redirect("/login")
|
return login_redirect_for_current_request()
|
||||||
return func(*args, **kwargs)
|
return func(*args, **kwargs)
|
||||||
|
|
||||||
return wrapper # type: ignore[return-value]
|
return wrapper # type: ignore[return-value]
|
||||||
|
|
@ -56,7 +76,7 @@ def require_admin(func: F) -> F:
|
||||||
def wrapper(*args, **kwargs):
|
def wrapper(*args, **kwargs):
|
||||||
user = current_user()
|
user = current_user()
|
||||||
if user is None:
|
if user is None:
|
||||||
return redirect("/login")
|
return login_redirect_for_current_request()
|
||||||
if not user.admin:
|
if not user.admin:
|
||||||
abort(403)
|
abort(403)
|
||||||
return func(*args, **kwargs)
|
return func(*args, **kwargs)
|
||||||
|
|
|
||||||
|
|
@ -1,9 +1,9 @@
|
||||||
import time
|
import time
|
||||||
|
|
||||||
from flask import Blueprint, Response, request, redirect
|
from flask import Blueprint, Response, redirect, render_template, request
|
||||||
from sqlalchemy import select
|
from sqlalchemy import select
|
||||||
|
|
||||||
from l4d2web.auth import hash_password, login_user, logout_user, verify_password
|
from l4d2web.auth import is_safe_next, login_user, logout_user, verify_password
|
||||||
from l4d2web.db import session_scope
|
from l4d2web.db import session_scope
|
||||||
from l4d2web.models import User
|
from l4d2web.models import User
|
||||||
|
|
||||||
|
|
@ -29,33 +29,10 @@ def is_login_rate_limited(remote_addr: str) -> bool:
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
@bp.get("/signup")
|
|
||||||
def signup_form() -> Response:
|
|
||||||
return Response("signup", mimetype="text/plain")
|
|
||||||
|
|
||||||
|
|
||||||
@bp.post("/signup")
|
|
||||||
def signup() -> Response:
|
|
||||||
username = request.form.get("username", "").strip()
|
|
||||||
password = request.form.get("password", "")
|
|
||||||
if not username or not password:
|
|
||||||
return Response("missing credentials", status=400)
|
|
||||||
|
|
||||||
with session_scope() as db:
|
|
||||||
existing = db.scalar(select(User).where(User.username == username))
|
|
||||||
if existing is not None:
|
|
||||||
return Response("username already exists", status=409)
|
|
||||||
user = User(username=username, password_digest=hash_password(password), admin=False)
|
|
||||||
db.add(user)
|
|
||||||
db.flush()
|
|
||||||
login_user(user.id)
|
|
||||||
|
|
||||||
return redirect("/dashboard")
|
|
||||||
|
|
||||||
|
|
||||||
@bp.get("/login")
|
@bp.get("/login")
|
||||||
def login_form() -> Response:
|
def login_form() -> str:
|
||||||
return Response("login", mimetype="text/plain")
|
next_target = request.args.get("next", "")
|
||||||
|
return render_template("login.html", next_target=next_target if is_safe_next(next_target) else "")
|
||||||
|
|
||||||
|
|
||||||
@bp.post("/login")
|
@bp.post("/login")
|
||||||
|
|
@ -72,7 +49,8 @@ def login() -> Response:
|
||||||
return Response("invalid credentials", status=401)
|
return Response("invalid credentials", status=401)
|
||||||
login_user(user.id)
|
login_user(user.id)
|
||||||
LOGIN_ATTEMPTS_BY_IP.pop(remote_addr, None)
|
LOGIN_ATTEMPTS_BY_IP.pop(remote_addr, None)
|
||||||
return redirect("/dashboard")
|
next_target = request.form.get("next", "")
|
||||||
|
return redirect(next_target if is_safe_next(next_target) else "/dashboard")
|
||||||
|
|
||||||
|
|
||||||
@bp.post("/logout")
|
@bp.post("/logout")
|
||||||
|
|
|
||||||
|
|
@ -35,6 +35,30 @@ textarea {
|
||||||
font: inherit;
|
font: inherit;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
label {
|
||||||
|
display: grid;
|
||||||
|
gap: var(--space-xs);
|
||||||
|
}
|
||||||
|
|
||||||
|
input,
|
||||||
|
select,
|
||||||
|
textarea {
|
||||||
|
background: var(--color-surface);
|
||||||
|
border: var(--line);
|
||||||
|
border-radius: var(--radius-s);
|
||||||
|
color: var(--color-text);
|
||||||
|
padding: var(--space-s) var(--space-m);
|
||||||
|
}
|
||||||
|
|
||||||
|
input:focus,
|
||||||
|
select:focus,
|
||||||
|
textarea:focus,
|
||||||
|
button:focus-visible,
|
||||||
|
a:focus-visible {
|
||||||
|
outline: 2px solid var(--color-focus);
|
||||||
|
outline-offset: 2px;
|
||||||
|
}
|
||||||
|
|
||||||
button {
|
button {
|
||||||
background: var(--color-primary);
|
background: var(--color-primary);
|
||||||
border: none;
|
border: none;
|
||||||
|
|
@ -60,3 +84,7 @@ button.danger {
|
||||||
.inline-form {
|
.inline-form {
|
||||||
display: inline;
|
display: inline;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
.auth-panel {
|
||||||
|
max-width: 28rem;
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -14,10 +14,12 @@
|
||||||
<header class="site-header">
|
<header class="site-header">
|
||||||
<div class="site-header-inner">
|
<div class="site-header-inner">
|
||||||
<nav class="primary-nav" aria-label="Main navigation">
|
<nav class="primary-nav" aria-label="Main navigation">
|
||||||
<a class="brand" href="/dashboard">left4me</a>
|
<a class="brand" href="{{ '/dashboard' if g.user else '/login' }}">left4me</a>
|
||||||
|
{% if g.user %}
|
||||||
<a href="/servers">servers</a>
|
<a href="/servers">servers</a>
|
||||||
<a href="/blueprints">blueprints</a>
|
<a href="/blueprints">blueprints</a>
|
||||||
<a href="/overlays">overlays</a>
|
<a href="/overlays">overlays</a>
|
||||||
|
{% endif %}
|
||||||
</nav>
|
</nav>
|
||||||
{% if g.user %}
|
{% if g.user %}
|
||||||
<nav class="account-nav" aria-label="Account navigation">
|
<nav class="account-nav" aria-label="Account navigation">
|
||||||
|
|
|
||||||
23
l4d2web/templates/login.html
Normal file
23
l4d2web/templates/login.html
Normal file
|
|
@ -0,0 +1,23 @@
|
||||||
|
{% extends "base.html" %}
|
||||||
|
|
||||||
|
{% block title %}Log In | left4me{% endblock %}
|
||||||
|
|
||||||
|
{% block content %}
|
||||||
|
<section class="panel auth-panel">
|
||||||
|
<h1>Log in to left4me</h1>
|
||||||
|
<p class="muted">Use your local account to manage Left 4 Dead 2 servers.</p>
|
||||||
|
|
||||||
|
<form method="post" action="/login" class="stack">
|
||||||
|
{% if next_target %}<input type="hidden" name="next" value="{{ next_target }}">{% endif %}
|
||||||
|
<label>
|
||||||
|
Username
|
||||||
|
<input name="username" autocomplete="username" required autofocus>
|
||||||
|
</label>
|
||||||
|
<label>
|
||||||
|
Password
|
||||||
|
<input name="password" type="password" autocomplete="current-password" required>
|
||||||
|
</label>
|
||||||
|
<button type="submit">log in</button>
|
||||||
|
</form>
|
||||||
|
</section>
|
||||||
|
{% endblock %}
|
||||||
|
|
@ -30,9 +30,46 @@ def seed_user(tmp_path, monkeypatch):
|
||||||
return user_id
|
return user_id
|
||||||
|
|
||||||
|
|
||||||
def test_public_signup(client) -> None:
|
def test_login_page_renders_form(client) -> None:
|
||||||
response = client.post("/signup", data={"username": "alice", "password": "secret"})
|
response = client.get("/login")
|
||||||
|
text = response.get_data(as_text=True)
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert '<form method="post" action="/login"' in text
|
||||||
|
assert 'name="username"' in text
|
||||||
|
assert 'name="password"' in text
|
||||||
|
assert "signup" not in text.lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_signup_routes_are_gone(client) -> None:
|
||||||
|
assert client.get("/signup").status_code == 404
|
||||||
|
assert client.post("/signup", data={"username": "alice", "password": "secret"}).status_code == 404
|
||||||
|
|
||||||
|
|
||||||
|
def test_login_redirects_to_safe_next(client) -> None:
|
||||||
|
with session_scope() as session:
|
||||||
|
session.add(User(username="alice", password_digest=hash_password("secret"), admin=False))
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/login",
|
||||||
|
data={"username": "alice", "password": "secret", "next": "/servers"},
|
||||||
|
)
|
||||||
|
|
||||||
assert response.status_code == 302
|
assert response.status_code == 302
|
||||||
|
assert response.headers["Location"].endswith("/servers")
|
||||||
|
|
||||||
|
|
||||||
|
def test_login_ignores_unsafe_next(client) -> None:
|
||||||
|
with session_scope() as session:
|
||||||
|
session.add(User(username="alice", password_digest=hash_password("secret"), admin=False))
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/login",
|
||||||
|
data={"username": "alice", "password": "secret", "next": "https://example.com/steal"},
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 302
|
||||||
|
assert response.headers["Location"].endswith("/dashboard")
|
||||||
|
|
||||||
|
|
||||||
def test_login_sets_session(client) -> None:
|
def test_login_sets_session(client) -> None:
|
||||||
|
|
|
||||||
|
|
@ -173,6 +173,44 @@ def test_non_admin_cannot_open_admin_pages(auth_client_with_server) -> None:
|
||||||
assert auth_client_with_server.get("/admin/jobs").status_code == 403
|
assert auth_client_with_server.get("/admin/jobs").status_code == 403
|
||||||
|
|
||||||
|
|
||||||
|
def test_anonymous_protected_page_redirects_to_login_with_next(tmp_path, monkeypatch) -> None:
|
||||||
|
db_url = f"sqlite:///{tmp_path/'anonymous-pages.db'}"
|
||||||
|
monkeypatch.setenv("DATABASE_URL", db_url)
|
||||||
|
app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
|
||||||
|
init_db()
|
||||||
|
client = app.test_client()
|
||||||
|
|
||||||
|
response = client.get("/servers")
|
||||||
|
|
||||||
|
assert response.status_code == 302
|
||||||
|
assert response.headers["Location"].endswith("/login?next=/servers")
|
||||||
|
|
||||||
|
|
||||||
|
def test_root_redirects_by_auth_state(tmp_path, monkeypatch) -> None:
|
||||||
|
db_url = f"sqlite:///{tmp_path/'root-redirect.db'}"
|
||||||
|
monkeypatch.setenv("DATABASE_URL", db_url)
|
||||||
|
app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
|
||||||
|
init_db()
|
||||||
|
client = app.test_client()
|
||||||
|
|
||||||
|
anonymous_response = client.get("/")
|
||||||
|
assert anonymous_response.status_code == 302
|
||||||
|
assert anonymous_response.headers["Location"].endswith("/login")
|
||||||
|
|
||||||
|
with session_scope() as session:
|
||||||
|
user = User(username="alice", password_digest=hash_password("secret"), admin=False)
|
||||||
|
session.add(user)
|
||||||
|
session.flush()
|
||||||
|
user_id = user.id
|
||||||
|
|
||||||
|
with client.session_transaction() as sess:
|
||||||
|
sess["user_id"] = user_id
|
||||||
|
|
||||||
|
logged_in_response = client.get("/")
|
||||||
|
assert logged_in_response.status_code == 302
|
||||||
|
assert logged_in_response.headers["Location"].endswith("/dashboard")
|
||||||
|
|
||||||
|
|
||||||
def test_blueprint_page_private_visibility(user_client_and_other_blueprint) -> None:
|
def test_blueprint_page_private_visibility(user_client_and_other_blueprint) -> None:
|
||||||
client, blueprint_id = user_client_and_other_blueprint
|
client, blueprint_id = user_client_and_other_blueprint
|
||||||
response = client.get(f"/blueprints/{blueprint_id}")
|
response = client.get(f"/blueprints/{blueprint_id}")
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue