me/pi-ku
1
0
Fork 0
mirror of https://github.com/ramvignesh-b/pi-ku.git synced 2026-05-04 08:56:52 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

refactor: clean up scaffolding backend

Browse Source
This commit is contained in:
ramvignesh-b
2026-04-16 03:30:42 +05:30
parent e8dac65468
commit cc8e3e4e4e
16 changed files with 174 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/config/settings.py
+20 -32
View File
@@ -35,27 +35,25 @@ SECRET_KEY = env("SECRET_KEY")
# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = env("DEBUG")
ALLOWED_HOSTS = []
ALLOWED_HOSTS = env.list("ALLOWED_HOSTS") or []
# Application definition
INSTALLED_APPS = [
"django.contrib.admin",
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.sessions",
"django.contrib.messages",
"django.contrib.staticfiles",
"rest_framework", # for API
"corsheaders", # for API and Frontend connect
"users", # custom user app
"letters", # letters app
"rest_framework",
"corsheaders",
"users",
"letters",
]
MIDDLEWARE = [
"corsheaders.middleware.CorsMiddleware", # allow frontend to connect
"django.middleware.security.SecurityMiddleware",
"corsheaders.middleware.CorsMiddleware",
"django.contrib.sessions.middleware.SessionMiddleware",
"django.middleware.common.CommonMiddleware",
"django.middleware.csrf.CsrfViewMiddleware",
@@ -66,21 +64,6 @@ MIDDLEWARE = [
ROOT_URLCONF = "config.urls"
TEMPLATES = [
{
"BACKEND": "django.template.backends.django.DjangoTemplates",
"DIRS": [],
"APP_DIRS": True,
"OPTIONS": {
"context_processors": [
"django.template.context_processors.request",
"django.contrib.auth.context_processors.auth",
"django.contrib.messages.context_processors.messages",
],
},
},
]
WSGI_APPLICATION = "config.wsgi.application"
@@ -99,7 +82,7 @@ DATABASES = {
}
CORS_ALLOWED_ORIGINS = env.list("CORS_ALLOWED_ORIGINS")
CORS_ALLOW_CREDENTIALS = True # allow cookies with frontend
CORS_ALLOW_CREDENTIALS = True
AUTH_USER_MODEL = "users.User"
@@ -107,27 +90,32 @@ REST_FRAMEWORK = {
"DEFAULT_AUTHENTICATION_CLASSES": ("rest_framework_simplejwt.authentication.JWTAuthentication",),
"DEFAULT_PERMISSION_CLASSES": ("rest_framework.permissions.IsAuthenticated",),
}
SIMPLE_JWT = {
"ACCESS_TOKEN_LIFETIME": timedelta(minutes=30),
# "ACCESS_TOKEN_LIFETIME": timedelta(seconds=10), # lazy testing
"REFRESH_TOKEN_LIFETIME": timedelta(days=1),
"ROTATE_REFRESH_TOKENS": True,
"BLACKLIST_AFTER_ROTATION": True,
"AUTH_HEADER_TYPES": ("Bearer",),
"AUTH_TOKEN_CLASSES": ("rest_framework_simplejwt.tokens.AccessToken",),
"AUTH_COOKIE": "refresh_token",
"AUTH_COOKIE_DOMAIN": None,
"AUTH_COOKIE_SECURE": not DEBUG,
"AUTH_COOKIE_HTTPONLY": True,
"AUTH_COOKIE_SAMESITE": "Lax", # Allow cross-site for links from email. Otherwise we'd use strict
}
"""
NOTE: COOKIE_SAMESITE: Lax is used to allow cross-site redirection, like links from email.
"""
AUTH_COOKIE = {
"NAME": "refresh_token",
"DOMAIN": None,
"SECURE": not DEBUG,
"HTTPONLY": True,
"SAMESITE": "Lax",
}
# Email config
EMAIL_BACKEND = "django.core.mail.backends.smtp.EmailBackend"
EMAIL_HOST = env("EMAIL_HOST")
EMAIL_PORT = env("EMAIL_PORT")
EMAIL_USE_TLS = not DEBUG # false for local, true for production
EMAIL_USE_SSL = False # since we enforce TLS
EMAIL_USE_TLS = not DEBUG
EMAIL_HOST_USER = env("EMAIL_HOST_USER")
EMAIL_HOST_PASSWORD = env("EMAIL_HOST_PASSWORD")
FROM_EMAIL = env("FROM_EMAIL")
backend/config/urls.py
+3 -4
View File
@@ -17,14 +17,13 @@ Including another URLconf
from django.conf import settings
from django.conf.urls.static import static
from django.contrib import admin
from django.urls import include, path
urlpatterns = [
path("admin/", admin.site.urls),
path("api/auth/", include("users.urls")), # user related operations
path("api/letters/", include("letters.urls")), # letter related operations
path("api/auth/", include("users.urls")),
path("api/letters/", include("letters.urls")),
]
# HACK: allow django directory to serve media files. In prod, ideally we use different storage backends (s3).
if settings.DEBUG:
urlpatterns += static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)
backend/letters/admin.py
-1
View File
@@ -1 +0,0 @@
# Register your models here.
backend/letters/migrations/0007_alter_letter_public_id.py
+19
View File
@@ -0,0 +1,19 @@
# Generated by Django 6.0.4 on 2026-04-15 18:13
import uuid
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("letters", "0006_letterimage"),
]
operations = [
migrations.AlterField(
model_name="letter",
name="public_id",
field=models.UUIDField(default=uuid.uuid4, primary_key=True, serialize=False),
),
]
backend/letters/models.py
+8 -1
View File
@@ -31,7 +31,9 @@ class Letter(models.Model):
encrypted_dek = models.TextField(null=True, blank=True)
def clean(self):
# custom validation
"""
Performs custom validation logic for Letter status and type relations.
"""
super().clean()
if self.type == Letter.Type.VAULT and self.status == Letter.Status.SEALED and not self.unlock_at:
raise ValidationError("A sealed VAULT letter must have an unlock_date.")
@@ -41,6 +43,11 @@ class Letter(models.Model):
class LetterImage(models.Model):
"""
Creates one to many relationship between Letter and Image.
Stores the uploaded images to server directory after inserting into the DB.
"""
public_id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
letter = models.ForeignKey(Letter, on_delete=models.CASCADE, related_name="images")
file_name = models.CharField(max_length=255)
backend/letters/serializers.py
+9 -4
View File
@@ -1,8 +1,6 @@
from rest_framework import serializers
from letters.models import LetterImage
from .models import Letter
from letters.models import Letter, LetterImage
class LetterImageSerializer(serializers.ModelSerializer):
@@ -16,6 +14,10 @@ class LetterSerializer(serializers.ModelSerializer):
images = LetterImageSerializer(many=True, read_only=True)
class Meta:
"""
Specifies the public_id as editable field for the client to generate.
"""
model = Letter
fields = [
"public_id",
@@ -29,10 +31,13 @@ class LetterSerializer(serializers.ModelSerializer):
"created_at",
"updated_at",
"images",
] # user to be fetched from request
]
read_only_fields = ["created_at", "updated_at"]
def validate(self, data):
"""
Validates the requirmnt of DEK when encrypted content and metadata are stored.
"""
if (data.get("encrypted_content") or data.get("encrypted_metadata")) and not data.get("encrypted_dek"):
raise serializers.ValidationError(
"encrypted_dek is required when encrypted_content and encrypted_metadata are present"
backend/letters/tests.py
+46 -14
View File
@@ -15,7 +15,9 @@ class LetterModelTest(TestCase):
self.user = User.objects.create_user(email="test@pi-ku.app", password="password1234", full_name="Test User")
def test_create_letter_draft(self):
"""create a basic Letter model with required fields"""
"""
Test the Letter model is created with required fields and auto timestamps.
"""
letter = Letter.objects.create(user=self.user, type="KEPT", status="DRAFT")
self.assertEqual(letter.user, self.user)
@@ -28,20 +30,24 @@ class LetterModelTest(TestCase):
self.assertIsNone(letter.sealed_at)
self.assertIsNone(letter.opened_at)
self.assertIsNone(letter.burned_at)
# Verify timestamps are auto-added
self.assertIsNotNone(letter.created_at)
self.assertIsNotNone(letter.updated_at)
def test_vault_requires_unlock_date_when_sealed(self):
"""a sealed VAULT letter must have an unlock_date"""
"""
Test that a sealed VAULT letter cannot be created without an unlock_date
"""
from django.core.exceptions import ValidationError
letter = Letter(
user=self.user,
type=Letter.Type.VAULT,
status=Letter.Status.SEALED,
encrypted_content="enc_v1...",
encrypted_content="enc_content==",
encrypted_metadata="enc_meta==",
encrypted_dek="enc_dek==",
)
with self.assertRaises(ValidationError):
letter.full_clean()
@@ -53,7 +59,9 @@ class LetterAPITest(APITestCase):
self.url = "/api/letters/"
def test_create_draft_letter_api(self):
"""Test API can successfully create a basic draft letter."""
"""
Test that the API can successfully create a basic draft letter.
"""
payload = {
"public_id": "4281edcc-5459-4ff2-bb5e-669fb44e0757",
"type": "KEPT",
@@ -63,6 +71,7 @@ class LetterAPITest(APITestCase):
}
response = self.client.put(self.url + payload["public_id"] + "/", payload)
self.assertEqual(response.status_code, 201)
self.assertEqual(Letter.objects.count(), 1)
self.assertEqual(Letter.objects.get().status, "DRAFT")
@@ -70,7 +79,9 @@ class LetterAPITest(APITestCase):
self.assertEqual(Letter.objects.get().user, self.user)
def test_update_draft_letter_with_public_id(self):
"""Test API can successfully update an existing letter with new values."""
"""
Test API can successfully update an existing letter with new values.
"""
letter = Letter.objects.create(
user=self.user,
type="KEPT",
@@ -87,7 +98,9 @@ class LetterAPITest(APITestCase):
"encrypted_metadata": "enc_meta==",
"encrypted_dek": "enc_dek==",
}
response = self.client.put(self.url + letter.public_id + "/", payload)
self.assertEqual(response.status_code, 200)
self.assertEqual(Letter.objects.count(), 1)
self.assertEqual(Letter.objects.get().status, "DRAFT")
@@ -98,7 +111,9 @@ class LetterAPITest(APITestCase):
self.assertEqual(Letter.objects.get().encrypted_dek, "enc_dek==")
def test_sealed_letters_cannot_be_updated(self):
"""Test API returns 400 when trying to update an already sealed letter."""
"""
Test that the API returns 400 when a user tries to update an already sealed letter.
"""
letter = Letter.objects.create(
user=self.user,
type="KEPT",
@@ -115,14 +130,20 @@ class LetterAPITest(APITestCase):
"encrypted_metadata": "enc_meta==",
"encrypted_dek": "enc_dek==",
}
response = self.client.put(self.url + letter.public_id + "/", payload)
self.assertEqual(response.status_code, 400)
self.assertEqual(response.data, {"error": "Sealed letters cannot be modified."})
def test_encrypted_dek_is_required_when_storing_encrypted_content_and_metadata(self):
"""encrypted_dek is required when encrypted_content and encrypted_metadata are present"""
"""
Test that encrypted_dek is required when encrypted_content and encrypted_metadata are added to the letter.
"""
payload = {"type": "KEPT", "encrypted_content": "enc_xyz==", "encrypted_metadata": "enc_meta=="}
response = self.client.post(self.url, payload)
self.assertEqual(response.status_code, 400)
self.assertEqual(Letter.objects.count(), 0)
self.assertEqual(
@@ -131,13 +152,14 @@ class LetterAPITest(APITestCase):
)
def test_create_letter_with_images_api(self):
"""Test API can create a letter and attach encrypted images in one request"""
"""
Test that the API can create a letter and attach encrypted images in one request.
"""
from django.core.files.uploadedfile import SimpleUploadedFile
# Simulate local encryption files
# Simulate local files upload
image1 = SimpleUploadedFile("enc_img1.bin", b"encrypted_bytes_1", content_type="application/octet-stream")
image2 = SimpleUploadedFile("enc_img2.bin", b"encrypted_bytes_2", content_type="application/octet-stream")
payload = {
"public_id": "4281edcc-5459-4ff2-bb5e-669fb44e0757",
"type": "SENT",
@@ -159,6 +181,9 @@ class LetterAPITest(APITestCase):
self.assertTrue(default_storage.exists("encrypted-images/enc_img2.bin"))
def test_cleanup_images_when_letter_is_updated(self):
"""
Test that the old images are cleaned up when a letter is updated with new images.
"""
letter = Letter.objects.create(user=self.user, type="KEPT", status="DRAFT")
LetterImage.objects.create(letter=letter, file_name="old1.bin", file=ContentFile(b"data", name="old1.bin"))
LetterImage.objects.create(letter=letter, file_name="old2.bin", file=ContentFile(b"data", name="old2.bin"))
@@ -176,7 +201,7 @@ class LetterAPITest(APITestCase):
from django.core.files.storage import default_storage
# Verify that the old files are cleared from storage
# Verify that the old files are cleared from storage directory as well
self.assertTrue(LetterImage.objects.filter(file_name="new.bin").exists())
self.assertEqual(LetterImage.objects.count(), 1)
self.assertFalse(default_storage.exists("encrypted-images/old1.bin"))
@@ -190,20 +215,27 @@ class LetterImageModelTest(TestCase):
self.letter = Letter.objects.create(user=self.user, type="KEPT", status="DRAFT")
def test_create_letter_image(self):
"""Test images can be associated with a letter (many to 1)"""
"""
Test that images can be associated with a letter (many to 1).
"""
image_content = ContentFile(b"fake-encrypted-data", name="test_image.bin")
letter_image = LetterImage.objects.create(
letter=self.letter, file_name="encrypted_image.enc", file=image_content
)
self.assertEqual(letter_image.letter, self.letter)
self.assertTrue(letter_image.file.name.startswith("encrypted-images/"))
self.assertIsNotNone(letter_image.public_id)
def test_letter_cascade_deletes_images(self):
"""TTest when a letter is deleted, its encrypted images are also removed"""
"""
TTest that when a letter is deleted, its encrypted images are also removed.
"""
LetterImage.objects.create(
letter=self.letter, file_name="will_be_deleted.jpg", file=ContentFile(b"data", name="del.bin")
)
self.assertEqual(LetterImage.objects.count(), 1)
self.letter.delete()
self.assertEqual(LetterImage.objects.count(), 0)
backend/letters/views.py
+19 -15
View File
@@ -8,11 +8,12 @@ from letters.serializers import LetterSerializer
class LetterView(generics.ListCreateAPIView):
serializer_class = LetterSerializer
# enforce auth guard
permission_classes = [IsAuthenticated]
def get_queryset(self):
"""return only letters of the authenticated user"""
"""
Returns the letters of the authenticated user.
"""
return Letter.objects.filter(user=self.request.user)
@@ -21,33 +22,37 @@ class LetterDetailView(generics.RetrieveUpdateDestroyAPIView):
lookup_field = "public_id"
def get_permissions(self):
"""
Allow any letter GET requests for guest access and enforce authentication for other operations.
"""
if self.request.method == "GET":
return [AllowAny()]
return [IsAuthenticated()]
def get_queryset(self):
"""
Returns the letters of the authenticated user.
Guests can only see SEALED letters.
"""
if self.request.user.is_authenticated:
# author can see all their letters (DRAFT, SEALED, etc.)
return Letter.objects.filter(user=self.request.user)
# guests can ONLY see SEALED letters
return Letter.objects.filter(status=Letter.Status.SEALED)
def put(self, request, public_id):
# upsert: create if doesn't exist, else update
"""
Upserts letters: create if doesn't exist, else update.
Validates the payload data, cleans up old images, and returns the upserted data.
"""
letter, created = Letter.objects.get_or_create(public_id=public_id, user=request.user)
# check if already sealed
if not created and letter.status == Letter.Status.SEALED:
return Response({"error": "Sealed letters cannot be modified."}, status=400)
# request.data handles both JSON and Multipart automatically in DRF
serializer = self.get_serializer(letter, data=request.data, partial=True)
serializer.is_valid(raise_exception=True)
serializer.save()
write_serializer = self.get_serializer(letter, data=request.data, partial=True)
write_serializer.is_valid(raise_exception=True)
write_serializer.save()
# Note: image_files is a list of binary files in request.FILES
if "image_files" in request.FILES:
# Delete old image files from storage and database
for old_image in letter.images.all():
old_image.file.delete(save=False)
old_image.delete()
@@ -55,6 +60,5 @@ class LetterDetailView(generics.RetrieveUpdateDestroyAPIView):
for image_file in request.FILES.getlist("image_files"):
LetterImage.objects.create(letter=letter, file=image_file, file_name=image_file.name)
# Return fresh data including the new image URLs
serializer = self.get_serializer(letter)
return Response(serializer.data, status=201 if created else 200)
response_serializer = self.get_serializer(letter)
return Response(response_serializer.data, status=201 if created else 200)
backend/main.py
-6
View File
@@ -1,6 +0,0 @@
def main():
print("Hello from backend!")
if __name__ == "__main__":
main()
backend/users/admin.py
-1
View File
@@ -1 +0,0 @@
# Register your models here.
backend/users/migrations/0005_remove_user_kdf_salt.py
+16
View File
@@ -0,0 +1,16 @@
# Generated by Django 6.0.4 on 2026-04-15 18:13
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("users", "0004_alter_user_public_id"),
]
operations = [
migrations.RemoveField(
model_name="user",
name="kdf_salt",
),
]
backend/users/models.py
+4 -13
View File
@@ -7,13 +7,12 @@ from django.utils.translation import gettext_lazy as _
class CustomUserManager(BaseUserManager):
"""
General User Model
Creates and saves a User with email and password.
"""
def create_user(self, email, password=None, **extra_fields):
if not email:
raise ValueError(_("The Email must be set"))
# set default activation state as False to enforce email verification
extra_fields.setdefault("is_active", False)
email = self.normalize_email(email)
@@ -24,7 +23,7 @@ class CustomUserManager(BaseUserManager):
def create_superuser(self, email, password, **extra_fields):
"""
Admin Model
Creates a Superuser with email and password.
"""
extra_fields.update({"is_staff": True, "is_superuser": True, "is_active": True})
@@ -33,30 +32,22 @@ class CustomUserManager(BaseUserManager):
class User(AbstractUser):
"""
Database table structure.
Note: We use the default integer ID internally for database performance (joins/indexes)
but expose a random 'public_id' (UUID) in URLs and APIs for real privacy.
Creates a User with email as primary identifier.
Requires manual email verification for activation.
"""
public_id = models.UUIDField(default=uuid.uuid4, editable=False, unique=True, db_index=True)
# Reset default fields
username = None
first_name = None
last_name = None
full_name = models.CharField(max_length=100)
email = models.EmailField(_("email address"), unique=True)
# salt for client-side key derivation
kdf_salt = models.CharField(max_length=128, blank=True, null=True)
# Default is False to enforce email verification
is_active = models.BooleanField(default=False)
objects = CustomUserManager()
# Login uses email instead of username
USERNAME_FIELD = "email"
REQUIRED_FIELDS = []
backend/users/serializers.py
+8
View File
@@ -8,10 +8,18 @@ class UserSerializer(serializers.ModelSerializer):
password = serializers.CharField(write_only=True)
class Meta:
"""
Specifies the public_id as readonly for the system to auto generate
"""
model = User
fields = ("public_id", "email", "full_name", "password")
read_only_fields = ("public_id",)
def create(self, validated_data):
"""
Validates and creates a new user with the given data.
"""
user = User.objects.create_user(
email=validated_data["email"],
password=validated_data["password"],
backend/users/tests.py
+13 -9
View File
@@ -1,4 +1,3 @@
from django.conf import settings
from django.contrib.auth import get_user_model
from django.contrib.auth.tokens import default_token_generator
from django.urls import reverse
@@ -21,30 +20,35 @@ class AuthTests(APITestCase):
self.logout_url = reverse("logout")
def test_login_sets_secure_cookie(self):
"""
Tests if the Login API can generate access token and set secure cookie for refresh token.
"""
data = {"email": self.user.email, "password": self.password}
cookie_name = "refresh_token"
response = self.client.post(self.login_url, data)
cookie_name = settings.SIMPLE_JWT["AUTH_COOKIE"]
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertIn("access", response.data)
self.assertNotIn("refresh", response.data)
self.assertIn(cookie_name, response.cookies)
# verify the cookie has a value
self.assertTrue(response.cookies[cookie_name].value)
self.assertTrue(response.cookies[cookie_name].httponly)
self.assertEqual(response.cookies[cookie_name]["samesite"], "Lax")
class ActivationTests(APITestCase):
def test_user_activation(self):
# initial user state
user = User.objects.create_user(email="inactive@test.com", password="password1234", is_active=False)
# generate activation link
"""
Tests if the Activation API can activate an inactive user.
"""
user = User.objects.create_user(email="inactiveuser@test.com", password="password1234", is_active=False)
uidb64 = urlsafe_base64_encode(force_bytes(user.public_id))
token = default_token_generator.make_token(user)
# call activation url
activation_url = reverse("activate", kwargs={"uidb64": uidb64, "token": token})
response = self.client.get(activation_url)
user.refresh_from_db()
self.assertEqual(response.status_code, status.HTTP_200_OK)
# check user is activated
user.refresh_from_db()
self.assertTrue(user.is_active)
backend/users/utils.py
+5 -5
View File
@@ -27,12 +27,12 @@ def set_response_cookies(response, refresh_token):
if "refresh" in _response.data:
del _response.data["refresh"] # remove refresh token from response body
_response.set_cookie(
key=settings.SIMPLE_JWT["AUTH_COOKIE"],
key=settings.AUTH_COOKIE["NAME"],
value=refresh_token,
max_age=settings.SIMPLE_JWT["REFRESH_TOKEN_LIFETIME"].total_seconds(),
httponly=settings.SIMPLE_JWT["AUTH_COOKIE_HTTPONLY"],
secure=settings.SIMPLE_JWT["AUTH_COOKIE_SECURE"],
samesite=settings.SIMPLE_JWT["AUTH_COOKIE_SAMESITE"],
domain=settings.SIMPLE_JWT["AUTH_COOKIE_DOMAIN"],
httponly=settings.AUTH_COOKIE["HTTPONLY"],
secure=settings.AUTH_COOKIE["SECURE"],
samesite=settings.AUTH_COOKIE["SAMESITE"],
domain=settings.AUTH_COOKIE["DOMAIN"],
)
return _response
backend/users/views.py
+4 -4
View File
@@ -74,7 +74,7 @@ class RefreshTokenView(TokenRefreshView):
permission_classes = (permissions.AllowAny,)
def post(self, request, *args, **kwargs):
refresh_token = request.COOKIES.get(settings.SIMPLE_JWT["AUTH_COOKIE"])
refresh_token = request.COOKIES.get(settings.AUTH_COOKIE["NAME"])
if not refresh_token:
return Response({"detail": "Refresh token not found"}, status=status.HTTP_401_UNAUTHORIZED)
request.data["refresh"] = refresh_token
@@ -92,9 +92,9 @@ class LogoutView(generics.GenericAPIView):
response = Response({"detail": "Successfully logged out"}, status=status.HTTP_200_OK)
# Clear the secure cookie
response.delete_cookie(
key=settings.SIMPLE_JWT["AUTH_COOKIE"],
domain=settings.SIMPLE_JWT.get("AUTH_COOKIE_DOMAIN"),
samesite=settings.SIMPLE_JWT.get("AUTH_COOKIE_SAMESITE"),
key=settings.AUTH_COOKIE["NAME"],
domain=settings.AUTH_COOKIE.get("DOMAIN"),
samesite=settings.AUTH_COOKIE.get("SAMESITE"),
path="/",
)
return response