moved webhook handler to Public API + require superuser auth

Signed-off-by: Trey <[email protected]>

backend/api/base/modal.py

@@ -5,6 +5,7 @@
 from django.shortcuts import render
 
 from backend.api.public.permissions import SCOPE_DESCRIPTIONS
+from backend.api.public.models import APIAuthToken
 from backend.models import Client, Receipt, User
 from backend.models import Invoice
 from backend.models import QuotaLimit
@@ -150,6 +151,7 @@ def open_modal(request: WebRequest, modal_name, context_type=None, context_value
                 {"name": group, "description": perms["description"], "options": perms["options"]}
                 for group, perms in SCOPE_DESCRIPTIONS.items()
             ]
+            context["APIAuthToken_types"] = APIAuthToken.AdministratorServiceTypes
 
         if modal_name == "edit_team_member_permissions":
             team = request.user.logged_in_as_team

backend/api/public/authentication.py

@@ -15,6 +15,7 @@ def get_model(self) -> Type[APIAuthToken]:
         return APIAuthToken
 
     def authenticate_credentials(self, raw_key) -> tuple[User | Organization | None, APIAuthToken]:
+        print("is here")
         model = self.get_model()
 
         try:

backend/api/public/endpoints/webhooks/urls.py

@@ -0,0 +1,12 @@
+from django.urls import path
+from .webhook_task_queue_handler import webhook_task_queue_handler_view_endpoint
+
+urlpatterns = [
+    path(
+        "receive/global/",
+        webhook_task_queue_handler_view_endpoint,
+        name="receive_global",
+    ),
+]
+
+app_name = "webhooks"

backend/webhooks/webhook_task_queue_handler.py → backend/api/public/endpoints/webhooks/webhook_task_queue_handler.py

@@ -1,22 +1,28 @@
-from django.http import JsonResponse
-from django.views.decorators.csrf import csrf_exempt
+from rest_framework.response import Response
+
+from backend.api.public.models import APIAuthToken
 import json
-from django.views.decorators.http import require_POST
+from rest_framework.decorators import api_view
 
 from backend.service.asyn_tasks.tasks import Task
 
 
-@csrf_exempt
-@require_POST
+@api_view(["POST"])
 def webhook_task_queue_handler_view_endpoint(request):
-    print("TASK 5 - Webhook Callback")
+    token: APIAuthToken | None = request.auth
+
+    if not token:
+        return Response({"status": "error", "message": "No token found"}, status=500)
+
+    if not token.administrator_service_type == token.AdministratorServiceTypes.AWS_WEBHOOK_CALLBACK:
+        return Response({"status": "error", "message": "Invalid API key for this service"}, status=500)
+
     try:
-        data: dict = json.loads(request.body)
+        data: dict = request.data
         func_name: str = data.get("func_name")
         args: list = data.get("args", [])
         kwargs: dict = data.get("kwargs", {})
 
-        print(data)
         print(f"Function Name: {func_name}")
         print(f"Arguments: {args}")
         print(f"Keyword Arguments: {kwargs}")
@@ -34,8 +40,8 @@ def webhook_task_queue_handler_view_endpoint(request):
         # Handle the result (e.g., store it or log it)
         print(f"Webhook executed: {func_name} with result: {result}")
 
-        return JsonResponse({"status": "success", "result": result})
+        return Response({"status": "success", "result": result})
 
     except Exception as e:
         print(f"Error executing webhook task: {str(e)}")
-        return JsonResponse({"status": "error", "message": str(e)}, status=500)
+        return Response({"status": "error", "message": str(e)}, status=500)

backend/api/public/models.py

@@ -23,6 +23,12 @@ class APIAuthToken(OwnerBase):
     active = models.BooleanField("Active", default=True, help_text="If the key is active")
     scopes = models.JSONField("Scopes", default=list, help_text="List of permitted scopes")
 
+    class AdministratorServiceTypes(models.TextChoices):
+        AWS_WEBHOOK_CALLBACK = "aws_webhook_callback", "AWS Webhook Callback"
+        AWS_API_DESTINATION = "aws_api_destination", "AWS API Destination"
+
+    administrator_service_type = models.CharField("Administrator Service Type", max_length=64, blank=True, null=True)
+
     class Meta:
         verbose_name = "API Key"
         verbose_name_plural = "API Keys"

backend/api/public/urls.py

@@ -16,6 +16,7 @@
     path("currency/", include(CURRENCY_CONVERSION)),
     path("clients/", include("backend.api.public.endpoints.clients.urls")),
     path("invoices/", include("backend.api.public.endpoints.Invoices.urls")),
+    path("webhooks/", include("backend.api.public.endpoints.webhooks.urls")),
 ]
 
 app_name = "public"

backend/api/settings/api_keys.py

@@ -16,11 +16,20 @@ def generate_api_key_endpoint(request: WebRequest) -> HttpResponse:
     name = request.POST.get("name")
     expiry = request.POST.get("expiry")
     description = request.POST.get("description")
+    administrator_toggle = True if request.POST.get("administrator") == "on" else False
+    administrator_type = request.POST.get("administrator_type")
 
     permissions: list = get_permissions_from_request(request)
 
     key_obj, key_response = generate_public_api_key(
-        request, request.user.logged_in_as_team or request.user, name, permissions, expires=expiry, description=description
+        request,
+        request.user.logged_in_as_team or request.user,
+        name,
+        permissions,
+        expires=expiry,
+        description=description,
+        administrator_toggle=administrator_toggle,
+        administrator_type=administrator_type,
     )
 
     if not key_obj:

backend/api/urls.py

@@ -5,7 +5,6 @@
 
 urlpatterns = [
     path("base/", include("backend.api.base.urls")),
-    path("admin/", include("backend.api.admin.urls")),
     path("teams/", include("backend.api.teams.urls")),
     path("receipts/", include("backend.api.receipts.urls")),
     path("invoices/", include("backend.api.invoices.urls")),

backend/management/commands/generate_aws_scheduler_apikey.py

@@ -1,6 +1,6 @@
+import uuid
 from django.core.management.base import BaseCommand
-
-from backend.models import APIKey
+from backend.api.public.models import APIAuthToken
 
 
 class Command(BaseCommand):
@@ -9,24 +9,24 @@ class Command(BaseCommand):
     """
 
     def handle(self, *args, **kwargs):
-        token = APIKey.objects.create(service=APIKey.ServiceTypes.AWS_API_DESTINATION)
-        key = f"{token.id}:{token.key}"
-        token.hash()
+        token = APIAuthToken(service=APIAuthToken.AdministratorServiceTypes.AWS_API_DESTINATION, name=str(uuid.uuid4()))
+        raw_key: str = token.generate_key()
+        token.save()
 
         self.stdout.write(
             f"""
         NOTE: Keep this key secret. It is used to authenticate your API requests with the AWS EventBridge API.
 
-        Your API Key: {key}
+        Your API Key: {raw_key}
 
         To use this API Key for development you can use:
 
-        pulumi config set api_destination-api_key {key}
+        pulumi config set api_destination-api_key {raw_key}
         pulumi up
 
         If you would like to use it for production use:
         pulumi stack select production
-        pulumi config set api_destination-api_key {key}
+        pulumi config set api_destination-api_key {raw_key}
         pulumi up
         """
         )

backend/migrations/0047_apiauthtoken_administrator_service_type.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.0.7 on 2024-08-09 15:44
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("backend", "0046_rename_status_invoicereminder_boto_schedule_status_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="apiauthtoken",
+            name="administrator_service_type",
+            field=models.CharField(blank=True, max_length=64, null=True, verbose_name="Administrator Service Type"),
+        ),
+    ]

backend/service/api_keys/generate.py

@@ -6,7 +6,15 @@
 
 
 def generate_public_api_key(
-    request, owner: User | Organization, api_key_name: str | None, permissions: list, *, expires=None, description=None
+    request,
+    owner: User | Organization,
+    api_key_name: str | None,
+    permissions: list,
+    *,
+    expires=None,
+    description=None,
+    administrator_toggle: bool = False,
+    administrator_type: str | None = None
 ) -> tuple[APIAuthToken | None, str]:
     if not validate_name(api_key_name):
         return None, "Invalid key name"
@@ -23,7 +31,21 @@ def generate_public_api_key(
     if validate_scopes(permissions).failed or not has_permission_to_create(request, owner):
         return None, "Invalid permissions"
 
-    token = APIAuthToken(name=api_key_name, description=description, expires=expires, scopes=permissions)  # type: ignore[arg-type, misc]
+    administrator_service_type = None
+
+    if request.user.is_superuser:
+        if administrator_toggle:
+            if administrator_type not in [option[0] for option in APIAuthToken.AdministratorServiceTypes.choices]:
+                return None, "Invalid administration type"
+            administrator_service_type = administrator_type
+
+    token = APIAuthToken(
+        name=api_key_name,
+        description=description,
+        expires=expires,
+        scopes=permissions,
+        administrator_service_type=administrator_service_type,
+    )  # type: ignore[arg-type, misc]
 
     raw_key: str = token.generate_key()
 

backend/service/asyn_tasks/tasks.py

@@ -14,7 +14,7 @@ def __init__(self, queue_url=None):
         self.region_name = os.environ.get("AWS_REGION_NAME")
         self.aws_access_key_id = os.environ.get("AWS_ACCESS_KEY_ID")
         self.aws_secret_access_key = os.environ.get("AWS_ACCESS_KEY")
-        self.WEBHOOK_URL = os.environ.get("SITE_URL", default="http://127.0.0.1:8000") + reverse("webhooks:receive_global")
+        self.WEBHOOK_URL = os.environ.get("SITE_URL", default="http://127.0.0.1:8000") + reverse("api:public:webhooks:receive_global")
 
         if self.queue_url:
             self.sqs_client = boto3.client(

backend/service/invoices/recurring/schedules/date_handlers.py

@@ -27,7 +27,7 @@ def get_monthly_cron(day_of_month: int | None = None, date: Date | None = None)
 
 def get_weekly_cron(day_of_week: int | None = None) -> WeeklyCronServiceResponse:
     if isinstance(day_of_week, int) and 0 <= day_of_week <= 6:
-        cron_expression = f"0 7 ? * {day_of_week} *"
+        cron_expression = f"0 7 ? * {day_of_week + 1} *"
         return WeeklyCronServiceResponse(True, cron_expression)
     else:
         return WeeklyCronServiceResponse(False, "", "Invalid input for weekly cron")

backend/service/invoices/recurring/validate/frequencies.py

@@ -22,7 +22,7 @@ def validate_and_update_frequency(
     match frequency.lower():
         # region Weekly
         case "weekly":
-            if frequency_day_of_week not in [i for i in "0123456"]:
+            if frequency_day_of_week not in [i for i in "1234567"]:
                 return ValidateFrequencyServiceResponse(error_message="Please select a valid day of the week")
 
             invoice_set.frequency = InvoiceRecurringSet.Frequencies.WEEKLY

backend/service/webhooks/get_url.py

@@ -4,4 +4,4 @@
 
 
 def get_global_webhook_response_url():
-    return os.environ.get("SITE_URL", default="http://127.0.0.1:8000") + reverse("webhooks:receive_recurring_invoices")
+    return os.environ.get("SITE_URL", default="http://127.0.0.1:8000") + reverse("api:public:webhooks:receive_global")

backend/webhooks/urls.py

@@ -1,15 +1,13 @@
 from django.urls import path, include
 
 from backend.webhooks.invoices.recurring import handle_recurring_invoice_webhook_endpoint
-from backend.webhooks.webhook_task_queue_handler import webhook_task_queue_handler_view_endpoint
 
 # from backend.webhooks.invoices.schedules import receive_scheduled_invoice_schedule, receive_scheduled_invoice_reminder
 
 urlpatterns = [
     # path("schedules/receive/schedule/", receive_scheduled_invoice_schedule, name="receive_scheduled_invoice schedule"),
     # path("schedules/receive/reminder/", receive_scheduled_invoice_reminder, name="receive_scheduled_invoice reminder"),
     path("schedules/receive/recurring_invoices/", handle_recurring_invoice_webhook_endpoint, name="receive_recurring_invoices"),
-    path("receive/global/", webhook_task_queue_handler_view_endpoint, name="receive_global"),
 ]
 
 app_name = "webhooks"

frontend/templates/modals/generate_api_key.html

@@ -49,6 +49,28 @@
             <span class="label-text-alt text-error">Please enter a message between 3 and 255 characters.</span>
         </label>
     </div>
+    {% if request.user.is_superuser %}
+    <div class="form-control w-full">
+        <label class="label">
+            Administrator Connection?
+        </label>
+
+        <div class="grid grid-cols-8 gap-4 items-center">
+            <input id="modal_input-administrator"
+                   name="administrator"
+                   type="checkbox"
+                   class="peer checkbox checkbox-block checkbox-bordered">
+            <select name="administrator_type" class="select select-bordered col-span-7 w-full">
+                {% for option in APIAuthToken_types %}
+                    <option value="{{ option }}">{{ option.label }}</option>
+                {% endfor %}
+            </select>
+        </div>
+        <label class="label peer-[&amp;:not(:placeholder-shown):not(:focus):invalid]:block hidden">
+            <span class="label-text-alt text-error">Please enter a valid name for your key</span>
+        </label>
+    </div>
+    {% endif %}
     <div class="collapse collapse-arrow border select-bordered mt-4">
         <input type="checkbox" />
         <div class="collapse-title text-xl font-medium">Permissions</div>

frontend/templates/pages/settings/pages/account.html

@@ -75,6 +75,6 @@
 {#<div class="card p-6 mb-4">#}
 {#    {% include 'pages/settings/settings/sessions.html' %}#}
 {#</div>#}
-{% if request.user.is_superuser %}
-    <div class="card">{% include 'pages/settings/settings/admin.html' %}</div>
-{% endif %}
+{#{% if request.user.is_superuser %}#}
+{#    <div class="card">{% include 'pages/settings/settings/admin.html' %}</div>#}
+{#{% endif %}#}