212 lines
11 KiB
Python
212 lines
11 KiB
Python
from datetime import timedelta
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.test import Client
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
from apps.accounts.magic_access import MAGIC_LINK_SESSION_KEY
|
|
from apps.accounts.models import MagicLinkAccessLog, TicketMagicLink, User
|
|
from apps.attachments.models import Attachment, AttachmentPolicy
|
|
from apps.customers.models import CustomerProfile
|
|
from apps.messaging.models import Message
|
|
from apps.tickets.models import TicketCategory
|
|
from apps.tickets.services import create_ticket, reply_ticket
|
|
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
CONFIRMATION_MESSAGE = (
|
|
"شما با حساب دیگری وارد سامانه شده‌اید. برای مشاهده تیکت مرتبط با این لینک، "
|
|
"از حساب فعلی خارج شوید. پس از خروج، تیکت به‌صورت خودکار باز می‌شود و نیازی به ورود مجدد ندارید."
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def magic_case(settings, tmp_path):
|
|
settings.MEDIA_ROOT = tmp_path
|
|
category = TicketCategory.objects.create(title="لینک امن", slug="magic-confirmation")
|
|
owner = User.objects.create_user(username="link-owner", mobile="09121000101", account_type="CUSTOMER")
|
|
other = User.objects.create_user(username="link-other", mobile="09121000102", account_type="CUSTOMER")
|
|
CustomerProfile.objects.create(user=owner)
|
|
CustomerProfile.objects.create(user=other)
|
|
ticket = create_ticket(customer=owner, category=category, subject="موضوع محرمانه", body="پیام محرمانه")
|
|
sibling = create_ticket(customer=owner, category=category, subject="تیکت خارج از لینک", body="پیام دیگر")
|
|
token, link = TicketMagicLink.issue_with_record(ticket)
|
|
return owner, other, ticket, sibling, link, reverse("accounts:magic", args=[token])
|
|
|
|
|
|
def confirmation_data(user):
|
|
return {"action": "logout_and_open", "account_id": str(user.pk)}
|
|
|
|
|
|
@pytest.mark.parametrize("account_type", ["CUSTOMER", "STAFF"])
|
|
def test_mismatched_account_get_requires_confirmation_without_leaking_ticket(client, magic_case, account_type):
|
|
owner, other, ticket, sibling, link, url = magic_case
|
|
other.account_type = account_type
|
|
other.save(update_fields=["account_type"])
|
|
client.force_login(other)
|
|
response = client.get(url)
|
|
content = response.content.decode()
|
|
assert response.status_code == 200
|
|
assert CONFIRMATION_MESSAGE in content
|
|
assert "خروج از حساب و مشاهده تیکت" in content
|
|
assert "انصراف از مشاهده این تیکت/پیام" in content
|
|
assert ticket.subject not in content and "پیام محرمانه" not in content
|
|
assert owner.mobile not in content
|
|
assert client.session["_auth_user_id"] == str(other.pk)
|
|
assert MAGIC_LINK_SESSION_KEY not in client.session
|
|
assert MagicLinkAccessLog.objects.count() == 0
|
|
link.refresh_from_db()
|
|
assert link.view_count == 0 and link.used_at is None
|
|
assert "no-store" in response["Cache-Control"]
|
|
assert response["Referrer-Policy"] == "same-origin"
|
|
|
|
|
|
@pytest.mark.parametrize("account_type", ["CUSTOMER", "STAFF"])
|
|
def test_confirmed_logout_opens_only_linked_ticket_and_uses_owner_for_reply(client, magic_case, account_type):
|
|
owner, other, ticket, sibling, link, url = magic_case
|
|
other.account_type = account_type
|
|
other.save(update_fields=["account_type"])
|
|
client.force_login(other)
|
|
session = client.session
|
|
session["unrelated_account_data"] = "must disappear"
|
|
session.save()
|
|
old_session_key = session.session_key
|
|
response = client.post(url, confirmation_data(other), follow=True)
|
|
assert response.status_code == 200
|
|
assert response.redirect_chain == [(reverse("tickets:detail", args=[ticket.public_id]), 302)]
|
|
assert "پیام محرمانه" in response.content.decode()
|
|
assert "فقط به همین تیکت دسترسی دارید" in response.content.decode()
|
|
assert "_auth_user_id" not in client.session
|
|
assert "unrelated_account_data" not in client.session
|
|
assert client.session.session_key != old_session_key
|
|
assert client.session[MAGIC_LINK_SESSION_KEY] == link.pk
|
|
for destination in (
|
|
reverse("tickets:detail", args=[sibling.public_id]),
|
|
reverse("tickets:list"), reverse("tickets:new"), reverse("accounts:profile"), reverse("payments:home"),
|
|
):
|
|
assert client.get(destination).status_code == 302
|
|
for name in ("messages", "poll"):
|
|
assert client.get(reverse(f"tickets:{name}", args=[ticket.public_id])).status_code == 200
|
|
reply = client.post(reverse("tickets:reply", args=[ticket.public_id]), {"body": "پاسخ پس از خروج"})
|
|
assert reply.status_code == 302
|
|
assert Message.objects.get(ticket=ticket, body="پاسخ پس از خروج").author_id == owner.pk
|
|
link.refresh_from_db()
|
|
assert link.view_count == 1
|
|
assert MagicLinkAccessLog.objects.filter(magic_link=link).count() == 1
|
|
|
|
|
|
def test_cancel_preserves_current_account_and_does_not_consume_link(client, magic_case):
|
|
owner, other, ticket, sibling, link, url = magic_case
|
|
client.force_login(other)
|
|
client.get(url)
|
|
response = client.get(reverse("accounts:home"), follow=True)
|
|
assert response.status_code == 200
|
|
assert client.session["_auth_user_id"] == str(other.pk)
|
|
assert MAGIC_LINK_SESSION_KEY not in client.session
|
|
assert MagicLinkAccessLog.objects.count() == 0
|
|
|
|
|
|
@pytest.mark.parametrize("invalid_state", ["expired", "revoked", "inactive", "unknown"])
|
|
def test_link_revalidated_before_logout(client, magic_case, invalid_state):
|
|
owner, other, ticket, sibling, link, url = magic_case
|
|
client.force_login(other)
|
|
assert client.get(url).status_code == 200
|
|
if invalid_state == "expired":
|
|
link.expires_at = timezone.now() - timedelta(seconds=1)
|
|
link.save(update_fields=["expires_at"])
|
|
elif invalid_state == "revoked":
|
|
link.revoked_at = timezone.now()
|
|
link.save(update_fields=["revoked_at"])
|
|
elif invalid_state == "inactive":
|
|
owner.is_active = False
|
|
owner.save(update_fields=["is_active"])
|
|
else:
|
|
url = reverse("accounts:magic", args=["unknown-token"])
|
|
response = client.post(url, confirmation_data(other))
|
|
assert response.status_code == 302 and response.url == reverse("accounts:login")
|
|
assert client.session["_auth_user_id"] == str(other.pk)
|
|
assert MAGIC_LINK_SESSION_KEY not in client.session
|
|
assert MagicLinkAccessLog.objects.count() == 0
|
|
|
|
|
|
@pytest.mark.parametrize("payload", [{}, {"action": "other"}, {"action": "logout_and_open", "account_id": "wrong"}])
|
|
def test_unconfirmed_post_does_not_logout(client, magic_case, payload):
|
|
owner, other, ticket, sibling, link, url = magic_case
|
|
client.force_login(other)
|
|
assert client.post(url, payload).status_code == 200
|
|
assert client.session["_auth_user_id"] == str(other.pk)
|
|
assert MagicLinkAccessLog.objects.count() == 0
|
|
|
|
|
|
def test_confirmation_from_previous_account_cannot_logout_new_account(client, magic_case):
|
|
owner, other, ticket, sibling, link, url = magic_case
|
|
client.force_login(other)
|
|
client.get(url)
|
|
replacement = User.objects.create_user(username="replacement", mobile="09121000103", account_type="CUSTOMER")
|
|
client.force_login(replacement)
|
|
response = client.post(url, confirmation_data(other))
|
|
assert response.status_code == 200
|
|
assert client.session["_auth_user_id"] == str(replacement.pk)
|
|
assert MagicLinkAccessLog.objects.count() == 0
|
|
|
|
|
|
def test_logout_confirmation_requires_csrf_and_valid_post_succeeds(magic_case):
|
|
owner, other, ticket, sibling, link, url = magic_case
|
|
client = Client(enforce_csrf_checks=True)
|
|
client.force_login(other)
|
|
client.get(url)
|
|
assert client.post(url, confirmation_data(other)).status_code == 403
|
|
assert client.session["_auth_user_id"] == str(other.pk)
|
|
response = client.post(url, {**confirmation_data(other), "csrfmiddlewaretoken": client.cookies["csrftoken"].value})
|
|
assert response.status_code == 302 and response.url == reverse("tickets:detail", args=[ticket.public_id])
|
|
assert "_auth_user_id" not in client.session
|
|
|
|
|
|
@pytest.mark.parametrize("authenticated", [False, True])
|
|
def test_existing_anonymous_and_matching_account_flow_remains_direct(client, magic_case, authenticated):
|
|
owner, other, ticket, sibling, link, url = magic_case
|
|
if authenticated:
|
|
client.force_login(owner)
|
|
response = client.get(url)
|
|
assert response.status_code == 302 and response.url == reverse("tickets:detail", args=[ticket.public_id])
|
|
assert client.get(response.url).status_code == 200
|
|
assert "no-store" in response["Cache-Control"]
|
|
if authenticated:
|
|
assert client.session["_auth_user_id"] == str(owner.pk)
|
|
assert MAGIC_LINK_SESSION_KEY not in client.session
|
|
else:
|
|
assert "_auth_user_id" not in client.session
|
|
assert client.session[MAGIC_LINK_SESSION_KEY] == link.pk
|
|
|
|
|
|
def test_link_attachment_permissions_remain_scoped_after_logout(client, magic_case):
|
|
owner, other, ticket, sibling, link, url = magic_case
|
|
AttachmentPolicy.objects.create(attachment_type="PDF", allowed_extensions=[".pdf"], max_file_size=1024 * 1024)
|
|
file = SimpleUploadedFile("linked.pdf", b"%PDF-1.4 linked", content_type="application/pdf")
|
|
reply_ticket(ticket_id=ticket.pk, user=owner, body="پیوست مجاز", files=[file])
|
|
attachment = ticket.ticket_messages.get(body="پیوست مجاز").attachments.get()
|
|
internal = Message.objects.create(ticket=ticket, author=other, body="یادداشت داخلی", is_internal=True)
|
|
internal_attachment = Attachment.objects.get(pk=attachment.pk)
|
|
internal_attachment.pk = None
|
|
internal_attachment.public_id = uuid4()
|
|
internal_attachment.message = internal
|
|
internal_attachment.save()
|
|
client.force_login(other)
|
|
client.post(url, confirmation_data(other))
|
|
assert client.get(reverse("attachments:download", args=[attachment.public_id])).status_code == 200
|
|
assert client.get(reverse("attachments:download", args=[internal_attachment.public_id])).status_code == 403
|
|
attachment.message = sibling.ticket_messages.first()
|
|
attachment.save(update_fields=["message"])
|
|
assert client.get(reverse("attachments:download", args=[attachment.public_id])).status_code == 403
|
|
|
|
|
|
@pytest.mark.parametrize("method", ["put", "delete"])
|
|
def test_unsupported_method_never_logs_out(client, magic_case, method):
|
|
owner, other, ticket, sibling, link, url = magic_case
|
|
client.force_login(other)
|
|
assert getattr(client, method)(url).status_code == 405
|
|
assert client.session["_auth_user_id"] == str(other.pk)
|