InviteMe/tests/test_security.py

import io

import pytest
from PIL import Image
from sqlalchemy.exc import IntegrityError

from app import db
from app.models import AuditLog, CheckIn, Event, Guest, SmtpSettings, User, utcnow
from app.services.email_service import send_invitation
from app.tasks import send_event_invitations_in_context
from app.utils.paths import event_photo_folder
from config import Config


def _user(username, role):
    user = User(username=username, role=role, is_active=True)
    user.set_password("Valid-Password-123")
    db.session.add(user)
    db.session.flush()
    return user


def _guest(event, email="guest@example.com"):
    guest = Guest(
        event_id=event.id,
        department="IT",
        full_name="Тестовый Гость",
        email=email,
        photo_filename="guest.png",
    )
    db.session.add(guest)
    db.session.flush()
    return guest


def test_organizer_cannot_read_another_event_api(app, client):
    with app.app_context():
        owner = _user("owner", "organizer")
        other = _user("other", "organizer")
        own_event = Event(name="Own", created_by_id=owner.id)
        other_event = Event(name="Other", created_by_id=other.id)
        db.session.add_all([own_event, other_event])
        db.session.flush()
        _guest(own_event, "own@example.com")
        _guest(other_event, "other@example.com")
        db.session.commit()
        owner_id = owner.id
        owner_version = owner.session_version
        other_event_id = other_event.id

    with client.session_transaction() as session:
        session["user_id"] = owner_id
        session["role"] = "organizer"
        session["session_version"] = owner_version

    response = client.get(f"/api/guests/search?q=Тест&event_id={other_event_id}")
    assert response.status_code == 403


def test_private_photo_requires_event_access(app, client):
    with app.app_context():
        owner = _user("owner", "organizer")
        other = _user("other", "organizer")
        event = Event(name="Private", created_by_id=owner.id)
        db.session.add(event)
        db.session.flush()
        guest = _guest(event)
        folder = event_photo_folder(event.id)
        folder.mkdir(parents=True, exist_ok=True)
        Image.new("RGB", (10, 10), "white").save(folder / guest.photo_filename)
        db.session.commit()
        other_id = other.id
        other_version = other.session_version
        guest_id = guest.id

    with client.session_transaction() as session:
        session["user_id"] = other_id
        session["role"] = "organizer"
        session["session_version"] = other_version

    assert client.get(f"/media/guests/{guest_id}/photo").status_code == 403


def test_staff_access_requires_event_assignment(app, client):
    with app.app_context():
        admin = _user("admin", "admin")
        staff = _user("staff-assigned", "staff")
        event = Event(name="Assigned", created_by_id=admin.id)
        db.session.add(event)
        db.session.flush()
        guest = _guest(event)
        db.session.commit()
        staff_id = staff.id
        staff_version = staff.session_version
        event_id = event.id
        guest_id = guest.id

    with client.session_transaction() as session:
        session["user_id"] = staff_id
        session["role"] = "staff"
        session["session_version"] = staff_version

    assert client.get(f"/api/guests/search?q=Тест&event_id={event_id}").status_code == 403

    with app.app_context():
        event = db.session.get(Event, event_id)
        staff = db.session.get(User, staff_id)
        event.staff_members.append(staff)
        db.session.commit()

    response = client.get(f"/api/guests/search?q=Тест&event_id={event_id}")
    assert response.status_code == 200
    assert response.get_json()[0]["id"] == guest_id


def test_invalid_image_upload_is_not_saved(app, client):
    with app.app_context():
        organizer = _user("organizer", "organizer")
        event = Event(name="Event", created_by_id=organizer.id)
        db.session.add(event)
        db.session.commit()
        organizer_id = organizer.id
        organizer_version = organizer.session_version
        event_id = event.id

    with client.session_transaction() as session:
        session["user_id"] = organizer_id
        session["role"] = "organizer"
        session["session_version"] = organizer_version
        session["current_event_id"] = event_id

    response = client.post(
        "/admin/upload",
        data={"photo_files": (io.BytesIO(b"<script>alert(1)</script>"), "photo.jpg")},
        content_type="multipart/form-data",
    )
    assert response.status_code == 302
    with app.app_context():
        assert not (event_photo_folder(event_id) / "photo.jpg").exists()


def test_mail_password_is_encrypted_at_rest(app):
    with app.app_context():
        settings = SmtpSettings.get_settings()
        settings.set_mail_password("smtp-password")
        db.session.commit()
        raw_value = settings.mail_password
        assert raw_value.startswith("fernet:")
        assert "smtp-password" not in raw_value
        assert settings.get_mail_password() == "smtp-password"


def test_security_headers_are_present(client):
    response = client.get("/login")
    assert response.headers["X-Content-Type-Options"] == "nosniff"
    assert response.headers["X-Frame-Options"] == "DENY"
    assert "frame-ancestors 'none'" in response.headers["Content-Security-Policy"]


def test_web_setup_can_be_disabled(app, client):
    app.config["ALLOW_WEB_SETUP"] = False
    assert client.get("/setup").status_code == 404


def test_unsafe_production_config_is_rejected():
    class UnsafeProductionConfig(Config):
        IS_PRODUCTION = True
        SECRET_KEY = "change-me-in-production"
        CREDENTIAL_ENCRYPTION_KEY = "x" * 32
        BASE_URL = "https://events.example.com"
        TRUSTED_HOSTS = ["events.example.com"]
        SQLALCHEMY_DATABASE_URI = "postgresql+psycopg://example"
        RATELIMIT_STORAGE_URI = "redis://example"
        SESSION_COOKIE_SECURE = True
        ENFORCE_HTTPS = True
        ALLOW_WEB_SETUP = False

    with pytest.raises(RuntimeError, match="SECRET_KEY"):
        UnsafeProductionConfig.validate()


def test_checkin_api_requires_csrf_when_enabled(app, client):
    with app.app_context():
        staff = _user("staff", "staff")
        event = Event(name="Event", created_by_id=staff.id)
        db.session.add(event)
        db.session.flush()
        guest = _guest(event)
        db.session.commit()
        staff_id = staff.id
        staff_version = staff.session_version
        token = guest.token

    with client.session_transaction() as session:
        session["user_id"] = staff_id
        session["role"] = "staff"
        session["session_version"] = staff_version
    app.config["WTF_CSRF_ENABLED"] = True

    assert client.post(f"/api/check-in/{token}").status_code == 400


def test_admin_guest_search_and_pagination(app, client):
    with app.app_context():
        organizer = _user("search-owner", "organizer")
        event = Event(name="Search", created_by_id=organizer.id)
        db.session.add(event); db.session.flush()
        for number in range(55):
            guest = _guest(event, f"guest-{number:02d}@example.com")
            guest.full_name = f"Гость {number:02d}"
        db.session.commit()
        user_id, version, event_id = organizer.id, organizer.session_version, event.id
    with client.session_transaction() as session:
        session.update(user_id=user_id, role="organizer", session_version=version, current_event_id=event_id)
    assert client.get("/admin/guests").data.count(b"guest-") == 50
    assert client.get("/admin/guests?page=2").data.count(b"guest-") == 5
    assert b"guest-54@example.com" in client.get("/admin/guests?q=guest-54@example.com").data
    assert client.get("/admin/upload/template").data.startswith(b"PK")


def test_duplicate_checkin_is_rejected_and_can_be_undone(app, client):
    with app.app_context():
        admin = _user("gate-admin", "admin")
        staff = _user("gate-staff", "staff")
        event = Event(name="Gate", created_by_id=admin.id)
        event.staff_members.append(staff)
        db.session.add(event); db.session.flush()
        guest = _guest(event, "gate@example.com")
        db.session.commit()
        staff_id, version, token, guest_id = staff.id, staff.session_version, guest.token, guest.id
    with client.session_transaction() as session:
        session.update(user_id=staff_id, role="staff", session_version=version)
    assert client.post(f"/api/check-in/{token}").status_code == 200
    assert client.post(f"/api/check-in/{token}").status_code == 409
    assert client.post(f"/api/check-in/{token}/undo", json={"reason": "Ошибка оператора"}).status_code == 200
    with app.app_context():
        assert CheckIn.query.filter_by(guest_id=guest_id).count() == 0
        assert AuditLog.query.filter_by(action="check_in_undo", target_id=str(guest_id)).count() == 1


def test_database_rejects_duplicate_checkins(app):
    with app.app_context():
        owner = _user("constraint-owner", "admin")
        event = Event(name="Constraint", created_by_id=owner.id)
        db.session.add(event); db.session.flush()
        guest = _guest(event, "constraint@example.com")
        db.session.flush()
        db.session.add_all([CheckIn(guest_id=guest.id, event_id=event.id), CheckIn(guest_id=guest.id, event_id=event.id)])
        with pytest.raises(IntegrityError):
            db.session.commit()
        db.session.rollback()


def test_public_invitation_rsvp_calendar_and_ticket(app, client, monkeypatch, tmp_path):
    with app.app_context():
        owner = _user("rsvp-owner", "organizer")
        event = Event(
            name="RSVP Event", created_by_id=owner.id, event_date=utcnow(),
            event_end_date=utcnow(), rsvp_enabled=True,
        )
        db.session.add(event); db.session.flush()
        guest = _guest(event, "rsvp@example.com")
        db.session.commit()
        token, guest_id = guest.token, guest.id

    assert client.get(f"/invite/{token}").status_code == 200
    response = client.post(f"/invite/{token}", data={
        "rsvp_status": "confirmed",
        "dietary_requirements": "Без орехов",
        "rsvp_comment": "Буду вовремя",
    })
    assert response.status_code == 302
    calendar = client.get(f"/invite/{token}/calendar.ics")
    assert calendar.status_code == 200
    assert b"BEGIN:VCALENDAR" in calendar.data

    ticket_path = tmp_path / "ticket.png"
    Image.new("RGB", (10, 10), "white").save(ticket_path)
    monkeypatch.setattr("app.routes.generate_ticket_image", lambda *_args, **_kwargs: ticket_path)
    assert client.get(f"/invite/{token}/ticket").status_code == 200
    with app.app_context():
        updated = db.session.get(Guest, guest_id)
        assert updated.rsvp_status == "confirmed"
        assert updated.dietary_requirements == "Без орехов"


def test_invitation_delivery_and_worker_status(app, monkeypatch):
    with app.app_context():
        owner = _user("worker-owner", "organizer")
        event = Event(name="Worker", created_by_id=owner.id)
        db.session.add(event); db.session.flush()
        guest = _guest(event, "worker@example.com")
        db.session.commit()
        monkeypatch.setattr("app.services.email_service.generate_ticket_image", lambda *_args, **_kwargs: None)
        monkeypatch.setattr("app.services.email_service._send_email", lambda **_kwargs: None)
        result = send_event_invitations_in_context(event.id, owner.id)
        db.session.refresh(guest)
        assert result["sent"] == 1
        assert guest.invitation_status == "sent"
        assert guest.invitation_attempts == 1