Secure API for listing users and groups

app/.env.docker-example

@@ -1,5 +1,6 @@
 DEBUG=1
 SECRET_KEY=foo
+SECRET_API_SECRET=bar
 DJANGO_PORT=8000
 DJANGO_ALLOWED_HOSTS="localhost 127.0.0.1 [::1]"
 DJANGO_SUPERUSER_USERNAME=admin1111

app/core/api/secret_permissions.py

@@ -0,0 +1,59 @@
+import hashlib
+import hmac
+import time
+
+from django.conf import settings
+from rest_framework.exceptions import PermissionDenied
+from rest_framework.permissions import BasePermission
+
+from core.constants import message_invalid_signature
+
+
+class _ApiFields:
+    def __init__(self, api_key: str, timestamp: str, signature: str):
+        self.api_key = api_key
+        self.timestamp = timestamp
+        self.signature = signature
+
+
+def _get_request_api_fields(request) -> _ApiFields:
+    api_key = request.headers.get("X-API-Key", "")
+    timestamp = request.headers.get("X-API-Timestamp", "")
+    signature = request.headers.get("X-API-Signature", "")
+    return _ApiFields(api_key=api_key, timestamp=timestamp, signature=signature)
+
+
+def _is_expected_signature(api_key: str, timestamp: str, signature: str) -> bool:
+    current_timestamp = int(time.time())
+    request_timestamp = int(timestamp)
+    if abs(current_timestamp - request_timestamp) > 10:
+        return False
+
+    # Recreate the message and calculate the expected signature
+    message = f"{timestamp}{api_key}"
+    expected_signature = hmac.new(
+        settings.SECRET_API_KEY.encode("utf-8"), message.encode("utf-8"), hashlib.sha256
+    ).hexdigest()
+
+    # Use constant-time comparison to prevent timing attacks
+    return hmac.compare_digest(signature, expected_signature)
+
+
+def _has_expected_request_signature(request) -> bool:
+    api_fields = _get_request_api_fields(request)
+    return _is_expected_signature(
+        api_fields.api_key, api_fields.timestamp, api_fields.signature
+    )
+
+
+class HasValidSignature(BasePermission):
+    def has_permission(self, request, __view__) -> bool:
+        if not _has_expected_request_signature(request):
+            raise PermissionDenied(message_invalid_signature)
+        return True
+
+    def has_object_permission(self, request, __view__, __obj__) -> bool:
+        if not _has_expected_request_signature(request):
+            raise PermissionDenied(message_invalid_signature)
+        return True
+        return _has_expected_request_signature(request)

app/core/api/secret_views.py

@@ -0,0 +1,20 @@
+# from rest_framework import serializers as rest_serializers
+from rest_framework import viewsets
+
+from core.api.secret_permissions import HasValidSignature
+from core.api.serializers import UserSerializer
+from core.models import User
+
+
+class SecretUserViewSet(viewsets.ReadOnlyModelViewSet):
+    permission_classes = [HasValidSignature]
+    queryset = User.objects.all()
+    # when instantiated, get_serializer_context will be called
+    serializer_class = UserSerializer
+
+    # get_serializer_context called to set include_groups to True
+    # to include groups in the response
+    def get_serializer_context(self):
+        context = super().get_serializer_context()
+        context["include_groups"] = True
+        return context

app/core/api/serializers.py

@@ -1,3 +1,4 @@
+from django.contrib.auth.models import Group
 from rest_framework import serializers
 from timezone_field.rest_framework import TimeZoneSerializerField
 
@@ -37,10 +38,23 @@ class Meta:
         )
 
 
+class GroupSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = Group
+        fields = ("id", "name")
+
+
 class UserSerializer(serializers.ModelSerializer):
-    """Used to retrieve user info"""
+    """
+    Serializer for User model.
+
+    Parameters:
+    - include_groups (bool): Flag to include user groups in the serialized output.
+    """
 
     time_zone = TimeZoneSerializerField(use_pytz=False)
+    # see get_groups method
+    groups = serializers.SerializerMethodField()
 
     class Meta:
         model = User
@@ -64,6 +78,7 @@ class Meta:
             "phone",
             "texting_ok",
             "time_zone",
+            "groups",
         )
         read_only_fields = (
             "uuid",
@@ -73,6 +88,12 @@ class Meta:
             "email",
         )
 
+    def get_groups(self, obj):
+        include_groups = self.context.get("include_groups", False)
+        if include_groups:
+            return GroupSerializer(obj.groups.all(), many=True).data
+        return None
+
 
 class ProjectSerializer(serializers.ModelSerializer):
     """Used to retrieve project info"""

app/core/api/urls.py

@@ -1,6 +1,7 @@
 from django.urls import path
 from rest_framework import routers
 
+from .secret_views import SecretUserViewSet
 from .views import AffiliateViewSet
 from .views import AffiliationViewSet
 from .views import EventViewSet
@@ -34,6 +35,10 @@
 router.register(
     r"stack-element-types", StackElementTypeViewSet, basename="stack-element-type"
 )
+router.register(
+    r"secret-api/getusers", SecretUserViewSet, basename="secret-api-getusers"
+)
+router.register(r"sdgs", SdgViewSet, basename="sdg")
 router.register(r"sdgs", SdgViewSet, basename="sdg")
 router.register(
     r"affiliations",

app/core/constants.py

@@ -0,0 +1 @@
+message_invalid_signature = "Invalid signature"

app/core/tests/test_secret_api.py

@@ -0,0 +1,95 @@
+import hashlib
+import hmac
+import time
+
+from django.contrib.auth.models import Group
+from django.urls import reverse
+from rest_framework import status
+from rest_framework.test import APIClient
+from rest_framework.test import APITestCase
+
+from core.models import User
+from peopledepot.settings import SECRET_API_KEY
+
+secret_url = reverse("secret-api-getusers-list")
+
+
+class SecretUserViewSetTests(APITestCase):
+    def setUp(self):
+        # Create a test user
+        self.client = APIClient()
+
+    def generate_signature(self, api_key, timestamp):
+        # Generate a valid signature
+        message = f"{timestamp}{api_key}"
+        return hmac.new(
+            SECRET_API_KEY.encode("utf-8"), message.encode(), hashlib.sha256
+        ).hexdigest()
+
+    def test_access_with_invalid_signature(self):
+        response = self.client.get(
+            secret_url,
+            HTTP_X_API_Key=SECRET_API_KEY,
+            HTTP_X_API_Timestamp=str(int(time.time())),
+            HTTP_X_API_Signature="invalidsignature",
+        )
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_access_with_invalid_timestamp(self):
+        invalid_timestamp = str(int(time.time()) - 20)  # Invalid timestamp (too old)
+        signature = self.generate_signature(SECRET_API_KEY, invalid_timestamp)
+        response = self.client.get(
+            secret_url,
+            HTTP_X_API_Key=SECRET_API_KEY,
+            HTTP_X_API_Timestamp=invalid_timestamp,
+            HTTP_X_API_Signature=signature,
+        )
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_succeeds_with_valid_signature_without_authentication(self):
+        timestamp = str(int(time.time()))
+        signature = self.generate_signature(SECRET_API_KEY, timestamp)
+        response = self.client.get(
+            secret_url,
+            HTTP_X_API_Key=SECRET_API_KEY,
+            HTTP_X_API_Timestamp=timestamp,
+            HTTP_X_API_Signature=signature,
+        )
+        assert response.status_code == status.HTTP_200_OK
+        self.client.force_authenticate(user=None)  # Log out the user
+
+    def test_succeeds_with_valid_signature_and_authentication(self):
+        user = User.objects.create_user(username="testuser", password="testpassword")
+        self.client.force_authenticate(user=user)
+        timestamp = str(int(time.time()))
+        signature = self.generate_signature(SECRET_API_KEY, timestamp)
+        response = self.client.get(
+            secret_url,
+            HTTP_X_API_Key=SECRET_API_KEY,
+            HTTP_X_API_Timestamp=timestamp,
+            HTTP_X_API_Signature=signature,
+        )
+        assert response.status_code == status.HTTP_200_OK
+        self.client.force_authenticate(user=None)  # Log out the user
+
+    def test_list_users(self):
+        user = User.objects.create_user(username="testuser", password="testpassword")
+        group = Group.objects.create(name="testgroup")
+        group2 = Group.objects.create(name="testgroup2")
+        # third group created to confirm len(user.groups) does not include group3
+        Group.objects.create(name="testgroup3")
+        user.groups.add(group)
+        user.groups.add(group2)
+
+        timestamp = str(int(time.time()))
+        signature = self.generate_signature(SECRET_API_KEY, timestamp)
+        response = self.client.get(
+            secret_url,
+            HTTP_X_API_Key=SECRET_API_KEY,
+            HTTP_X_API_Timestamp=timestamp,
+            HTTP_X_API_Signature=signature,
+        )
+        assert response.status_code == status.HTTP_200_OK
+
+        group_count = len(response.data[0]["groups"])
+        assert group_count == 2

app/peopledepot/settings.py

@@ -23,7 +23,10 @@
 # See https://docs.djangoproject.com/en/4.0/howto/deployment/checklist/
 
 # SECURITY WARNING: keep the secret key used in production secret!
+# SECRET_KEY is used internally by Django to encrypt session data
+# SECRET_API_KEY is used to authenticate a "secret" API endpoint
 SECRET_KEY = os.environ.get("SECRET_KEY")
+SECRET_API_KEY = os.environ.get("SECRET_API_KEY")
 
 DJANGO_SUPERUSER_USERNAME = os.environ.get("DJANGO_SUPERUSER_USERNAME")
 DJANGO_SUPERUSER_EMAIL = os.environ.get("DJANGO_SUPERUSER_EMAIL")

app/requirements.txt

@@ -61,6 +61,7 @@ platformdirs==4.2.0
     # via black
 pluggy==1.4.0
     # via pytest
+pre-commit==3.7.1
 psycopg2-binary==2.9.9
 pycodestyle==2.11.1
     # via flake8

scripts/start-local.sh

@@ -11,11 +11,6 @@ if [[ $PWD != *"app"* ]]; then
     }
 fi
 
-SCRIPT_DIR="$(dirname "$0")"
-"$SCRIPT_DIR"/loadenv.sh || {
-    echo "ERROR: loadenv.sh failed"
-    return 1
-}
 echo Admin user = "$DJANGO_SUPERUSER" email = "$DJANGO_SUPERUSER_EMAIL"
 if [[ $1 != "" ]]; then
     port=$1