intent/tests/test_hardening.py

"""Phase 5: rate limits, security headers, moderation, the admin's second factor, email."""

import io

import pytest
from django.core.exceptions import ValidationError
from django.core.management import CommandError, call_command
from django.urls import reverse
from django_otp.oath import totp
from django_otp.plugins.otp_totp.models import TOTPDevice

from intent.accounts.admin import IntentUserAdmin, orgs_left_without_owner
from intent.accounts.models import User
from intent.comments import services as comment_services
from intent.core import throttle as limits
from intent.features.services import create_feature
from intent.issues.services import create_issue


def test_comment_rate_limit(open_project, owner, outsider, monkeypatch):
    monkeypatch.setattr(limits, "COMMENTS", (2, 60))
    feature = create_feature(open_project, owner, title="F")
    thread = comment_services.start_feature_thread(feature, outsider, body="1")
    comment_services.reply(thread, outsider, body="2")
    with pytest.raises(ValidationError, match="too often"):
        comment_services.reply(thread, outsider, body="3")
    comment_services.reply(thread, owner, body="Others aren't affected")


def test_issue_rate_limit(open_project, outsider, monkeypatch):
    monkeypatch.setattr(limits, "ISSUES", (1, 60))
    create_issue(open_project, outsider, title="One")
    with pytest.raises(ValidationError, match="too often"):
        create_issue(open_project, outsider, title="Two")


def test_rate_limited_reply_keeps_the_draft(client, open_project, owner, outsider, monkeypatch):
    monkeypatch.setattr(limits, "COMMENTS", (1, 60))
    feature = create_feature(open_project, owner, title="F")
    thread = comment_services.start_feature_thread(feature, outsider, body="1")
    client.force_login(outsider)
    url = reverse(
        "comments:reply",
        kwargs={"org_slug": "catbriar", "project_slug": "intent", "thread_id": thread.pk},
    )
    response = client.post(url, {"body": "My long reply"}, headers={"HX-Request": "true"})
    assert b"too often" in response.content and b"My long reply" in response.content


def test_security_headers(client, db):
    response = client.get(reverse("home"))
    csp = response["Content-Security-Policy"]
    assert "script-src 'self'" in csp and "'unsafe-inline'" not in csp
    assert "object-src 'none'" in csp and "frame-ancestors 'self'" in csp
    assert "camera=()" in response["Permissions-Policy"]
    assert response["X-Content-Type-Options"] == "nosniff"
    assert response["Referrer-Policy"] == "same-origin"


def test_bucket_origin_is_allowed_for_images_and_frames(settings, client, db):
    from intent.core.middleware import _content_security_policy

    settings.CSP_BUCKET_ORIGIN = "https://intent-files.fly.storage.tigris.dev"
    csp = _content_security_policy()
    assert "img-src 'self' data: https://intent-files.fly.storage.tigris.dev" in csp
    assert "frame-src 'self' https://intent-files.fly.storage.tigris.dev" in csp


# Moderation


def test_deactivated_user_is_signed_out_and_content_stays(client, open_project, owner, outsider):
    issue = create_issue(open_project, outsider, title="Mine")
    client.force_login(outsider)
    assert client.get(reverse("invites:list")).status_code == 200
    outsider.is_active = False
    outsider.save()
    assert client.get(reverse("invites:list")).status_code == 302  # Treated as signed out.
    issue.refresh_from_db()
    assert issue.author == outsider


def test_deleted_users_content_shows_as_ghost(client, open_project, owner, outsider):
    issue = create_issue(open_project, outsider, title="Orphan")
    outsider.delete()
    client.force_login(owner)
    page = client.get(
        reverse(
            "issues:detail",
            kwargs={"org_slug": "catbriar", "project_slug": "intent", "number": issue.number},
        )
    ).content.decode()
    assert "Opened by ghost" in page


def test_ghost_and_official_names_are_reserved(client, db):
    response = client.post(
        reverse("account_signup"), {"email": "g@example.com", "username": "Ghost"}
    )
    assert b"reserved" in response.content
    assert not User.objects.filter(username__iexact="ghost").exists()


def test_last_owner_cannot_be_deleted_in_admin(rf, owner, org, make_user):
    admin_user = User.objects.create_superuser("sam", "sam@example.com", "pw-long-enough-1")
    request = rf.get("/")
    request.user = admin_user
    from django.contrib import admin

    model_admin = IntentUserAdmin(User, admin.site)
    assert [o.slug for o in orgs_left_without_owner([owner])] == ["catbriar"]
    assert not model_admin.has_delete_permission(request, owner)
    assert "delete_selected" not in model_admin.get_actions(request)
    second = make_user("otto")
    org.memberships.create(user=second, role="owner")
    assert model_admin.has_delete_permission(request, owner)


def test_deactivate_action_ends_sessions(client, outsider, rf):
    from django.contrib.sessions.models import Session

    client.force_login(outsider)
    assert Session.objects.count() == 1
    admin_user = User.objects.create_superuser("sam", "sam@example.com", "pw-long-enough-1")
    request = rf.post("/")
    request.user = admin_user
    request._messages = type("M", (), {"add": lambda *a, **k: None})()
    IntentUserAdmin(User, None).deactivate(request, User.objects.filter(pk=outsider.pk))
    outsider.refresh_from_db()
    assert not outsider.is_active and Session.objects.count() == 0


# Admin second factor


def test_admin_needs_a_verified_device(client, db):
    staff = User.objects.create_superuser("sam", "sam@example.com", "pw-long-enough-1")
    client.force_login(staff)  # Password alone, no device.
    response = client.get(reverse("admin:index"))
    assert response.status_code == 302 and "/admin/login/" in response["Location"]


def test_admin_login_with_password_and_code(client, db):
    staff = User.objects.create_superuser("sam", "sam@example.com", "pw-long-enough-1")
    device = TOTPDevice.objects.create(user=staff, name="Phone", confirmed=True)
    login = reverse("admin:login")
    bad = client.post(
        login, {"username": "sam", "password": "pw-long-enough-1", "otp_token": "000000"}
    )
    assert bad.status_code == 200 and not client.get(reverse("admin:index")).status_code == 200
    code = f"{totp(device.bin_key, device.step, device.t0, device.digits):06d}"
    client.post(
        login,
        {"username": "sam", "password": "pw-long-enough-1", "otp_device": device.persistent_id,
         "otp_token": code},
    )  # fmt: skip
    assert client.get(reverse("admin:index")).status_code == 200


def test_add_admin_totp_command(db, make_user):
    staff = User.objects.create_superuser("sam", "sam@example.com", "pw-long-enough-1")
    out = io.StringIO()
    call_command("add_admin_totp", "sam", stdout=out)
    assert "otpauth://totp/" in out.getvalue()
    assert TOTPDevice.objects.filter(user=staff, confirmed=True).count() == 1
    with pytest.raises(CommandError, match="already"):
        call_command("add_admin_totp", "sam", stdout=io.StringIO())
    call_command("add_admin_totp", "sam", "--replace", stdout=io.StringIO())
    assert TOTPDevice.objects.filter(user=staff).count() == 1
    with pytest.raises(CommandError, match="isn't staff"):
        call_command("add_admin_totp", make_user("ruth").username, stdout=io.StringIO())


# Email


def test_anymail_settings_come_from_the_environment(monkeypatch):
    import importlib

    import config.settings.base as base

    monkeypatch.setenv("ANYMAIL_POSTMARK_SERVER_TOKEN", "token-123")
    reloaded = importlib.reload(base)
    try:
        assert reloaded.ANYMAIL == {"POSTMARK_SERVER_TOKEN": "token-123"}
    finally:
        monkeypatch.delenv("ANYMAIL_POSTMARK_SERVER_TOKEN")
        importlib.reload(base)