mirror of
https://github.com/barkeser2002/offline-db.git
synced 2026-09-25 13:46:12 +03:00
Barış Keserandgoogle-labs-jules[bot]
93693981f9
This commit adds a `password` field to the `Room` model to support private watch party rooms. It updates the `RoomSerializer` to include the password (write-only) and a boolean indicator (`is_private`). Furthermore, it enforces password verification in both the REST API (`RoomViewSet.retrieve`) and the WebSocket layer (`WatchPartyConsumer.connect`). Corresponding security tests have been added. Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
104 lines
4.2 KiB
Python
104 lines
4.2 KiB
Python
from rest_framework.test import APITestCase
|
|
from django.urls import reverse
|
|
from django.contrib.auth import get_user_model
|
|
from content.models import Anime, Season, Episode
|
|
from apps.watchparty.models import Room
|
|
|
|
User = get_user_model()
|
|
|
|
class RoomViewSetSecurityTests(APITestCase):
|
|
def setUp(self):
|
|
# Create a host user
|
|
self.host_user = User.objects.create_user(username='host', password='password')
|
|
# Create an attacker user
|
|
self.attacker_user = User.objects.create_user(username='attacker', password='password')
|
|
|
|
# Create Anime, Season, Episode
|
|
self.anime = Anime.objects.create(title="Test Anime")
|
|
self.season = Season.objects.create(anime=self.anime, number=1)
|
|
self.episode = Episode.objects.create(season=self.season, number=1)
|
|
|
|
# Create a Room hosted by host_user
|
|
self.room = Room.objects.create(
|
|
host=self.host_user,
|
|
episode=self.episode,
|
|
is_active=True
|
|
)
|
|
|
|
def test_attacker_cannot_delete_room(self):
|
|
# Attacker logs in
|
|
self.client.login(username='attacker', password='password')
|
|
|
|
# Attacker tries to delete the host's room
|
|
url = reverse('room-detail', kwargs={'pk': self.room.uuid})
|
|
response = self.client.delete(url, secure=True)
|
|
|
|
# If it returns 204, the vulnerability exists
|
|
if response.status_code == 204:
|
|
self.fail("Vulnerability: Attacker was able to delete someone else's room")
|
|
self.assertIn(response.status_code, [403, 404])
|
|
|
|
def test_attacker_cannot_update_room(self):
|
|
# Attacker logs in
|
|
self.client.login(username='attacker', password='password')
|
|
|
|
# Attacker tries to update the host's room
|
|
url = reverse('room-detail', kwargs={'pk': self.room.uuid})
|
|
response = self.client.patch(url, {'is_active': False}, secure=True)
|
|
|
|
# If it returns 200 OK, the vulnerability exists
|
|
if response.status_code == 200:
|
|
self.fail("Vulnerability: Attacker was able to update someone else's room")
|
|
self.assertIn(response.status_code, [403, 404])
|
|
|
|
def test_host_can_update_and_delete_room(self):
|
|
self.client.login(username='host', password='password')
|
|
url = reverse('room-detail', kwargs={'pk': self.room.uuid})
|
|
|
|
# Host can update (this will set is_active to False)
|
|
# However, the viewset's queryset is `Room.objects.filter(is_active=True)`
|
|
# Which means after setting is_active=False, a delete will 404
|
|
# We'll update something else to keep it active, or just do delete first
|
|
|
|
# Actually, let's just create another room or just delete directly
|
|
response = self.client.delete(url, secure=True)
|
|
self.assertEqual(response.status_code, 204)
|
|
|
|
def test_private_room_access(self):
|
|
# Create a private room
|
|
private_room = Room.objects.create(
|
|
host=self.host_user,
|
|
episode=self.episode,
|
|
is_active=True,
|
|
password='secretpassword'
|
|
)
|
|
|
|
url = reverse('room-detail', kwargs={'pk': private_room.uuid})
|
|
|
|
# Unauthenticated user without password
|
|
response = self.client.get(url, secure=True)
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
# Unauthenticated user with incorrect password
|
|
response = self.client.get(f"{url}?password=wrong", secure=True)
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
# Unauthenticated user with correct password
|
|
response = self.client.get(f"{url}?password=secretpassword", secure=True)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
# Host without password
|
|
self.client.login(username='host', password='password')
|
|
response = self.client.get(url, secure=True)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.client.logout()
|
|
|
|
# Attacker without password
|
|
self.client.login(username='attacker', password='password')
|
|
response = self.client.get(url, secure=True)
|
|
self.assertEqual(response.status_code, 403)
|
|
|
|
# Attacker with correct password
|
|
response = self.client.get(f"{url}?password=secretpassword", secure=True)
|
|
self.assertEqual(response.status_code, 200)
|