feat: WIP - licensing

api/app/settings/common.py

@@ -106,6 +106,7 @@
     "organisations",
     "organisations.invites",
     "organisations.permissions",
+    "organisations.subscriptions.licensing",
     "projects",
     "sales_dashboard",
     "edge_api",
@@ -1271,3 +1272,6 @@
 # subscriptions created before this date full audit log and versioning
 # history.
 VERSIONING_RELEASE_DATE = env.date("VERSIONING_RELEASE_DATE", default=None)
+
+SUBSCRIPTION_LICENCE_PUBLIC_KEY = env.str("SUBSCRIPTION_LICENCE_PUBLIC_KEY", None)
+SUBSCRIPTION_LICENCE_PRIVATE_KEY = env.str("SUBSCRIPTION_LICENCE_PRIVATE_KEY", None)

api/app/settings/test.py

@@ -18,3 +18,46 @@
 RETRY_WEBHOOKS = True
 
 INFLUXDB_BUCKET = "test_bucket"
+
+SUBSCRIPTION_LICENCE_PRIVATE_KEY = """
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC0o6Q+J6ArJZ2x
+RyZQ5e9ue6dB4bgH7I7DYYb9t9eIb55z0vZZWVLLmIr+ngCfCxIePqCclrAen9gr
+rCRhyAXD+XZYjRP0w2wlqA367HJXbti1adXnQnM4QXITNJhRnGoqiRVx7vQ/Klup
++yMBJOU4IkkSsQaAgp0eTdPlGlA+KAfCH39rsqIHNXuS1qfspI2RyaR6130NvR6D
+4p07XJls1AYOs8xphdWl8b4hzbJTvC0IqRhvX+z4kEyQjprdcfwOG4qrqtIb4asm
+21imOtE8CGRvHUl/cV+1l/hgv1fdbeCFzM89q16Z/KXIAWJMYfkWuOWGVEmf7yjB
+9aMrfM3fAgMBAAECggEAKqGwQocBkw1GoS8kiNUrY8zFFZRa5Wvb6ZqbzEdWE7oc
+EEPKph2hn7E5pIvPo7luJjsrlqktmZyp3Oy8jWMykSTP3Gg3PH3eiSiXXA/vkFj1
+xiLbO8AAB1fSv1ubUy9yEuXVbNUzSbEKfxxpD30Qp+XXjxS+bxfkUuGVT62dIH3V
+j251CEsCIZzwOriGP52OKK5HR24Y9/c+uGLu1CLY6qdrMgWAXTYqEoUw7ku8Sm8B
+o6fuu9i0mAEJUl6qcVz3yH0QYe9pM6jDQ9oZeSkVYyspCwysTVs2jsCTMYUpK/kD
+WU9sniHRgly3C9Ge3PrE4qUeMRTNk0Vd4RETtznwqQKBgQD3UiJ11FUd3g1FXL9N
+iLOIayACrX37cBuc8M5iAzasNDjSVNoCrQun1091xzYK6/F7As4PtH1YUaDgXdBp
+efHHl3DPTFkeztPMOKOd8tCpLai/23sbCBNc51x0LCuWWnNaAuidId1rpvFq7AMJ
+jE4HPJkoL6udOzlKUHebp02ILQKBgQC6+nT2A+AZeheUiw2wBl0BRitQCxA+TN+L
+vkAwLa0u/OqeNc8W50lybzHCS9nEVZ0Lp3Qk9Cl++X/k5o6k2byqJxtmMvLGqjjw
+UNuZWHSoUzfdzs8yBjroLM4HsBgbEaG9E2e2zuqKBvwLqZ3fv/fXvmJDIu+aCWXC
+ADtlrAvJuwKBgQDq+CW1PJ4BWk3RcGRwDUhEe0JWSO5ATCpv2Hi7tcHjqVmyutrF
+YBKKy4y6oSE/DxrFe8y6LwhHOIZXo8m17B1BOyf6StcA5g9jHwyTq3WCxdZlMOis
+red3hHfaB30Bw72D7u+BGgN7m4gRxVi9YYdgaLo569Bn+TRc3kZEo5aNoQKBgH7z
+aJBU50ZFCFeZ5iw61dD0pJnPOTMjnLBT917+1FRP8riCzl29obep2b4TJANTIbL0
++j3Q7Y/BtV1kUTuKfreEn+zO8NmEX+6C5+cBEQvsnMTkEvfjFQHo0eaUYHmYihlH
+YKbVbJdU0LLWclOmEpAQOsVcphQPB2EmKS4KF2LbAoGAOqVsQg61S1u7s4NF4JCN
+EiJvBDjjwTycNCmhY7bV1R7LX+Qk/Mq9fgK3yccKV/Bl69C9Fmeopivbu20urNhn
+q/sgOPDK0zJUSVh76gFon1gx7OfaHV31TrvIl0T7WnyfDvAv20F+dmmXkjnPBNNm
+dXzo4kXwDOlWCJI8VhYfH/0=
+-----END PRIVATE KEY-----
+"""
+
+SUBSCRIPTION_LICENCE_PUBLIC_KEY = """
+-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAtKOkPiegKyWdsUcmUOXv
+bnunQeG4B+yOw2GG/bfXiG+ec9L2WVlSy5iK/p4AnwsSHj6gnJawHp/YK6wkYcgF
+w/l2WI0T9MNsJagN+uxyV27YtWnV50JzOEFyEzSYUZxqKokVce70PypbqfsjASTl
+OCJJErEGgIKdHk3T5RpQPigHwh9/a7KiBzV7ktan7KSNkcmketd9Db0eg+KdO1yZ
+bNQGDrPMaYXVpfG+Ic2yU7wtCKkYb1/s+JBMkI6a3XH8DhuKq6rSG+GrJttYpjrR
+PAhkbx1Jf3FftZf4YL9X3W3ghczPPatemfylyAFiTGH5FrjlhlRJn+8owfWjK3zN
+3wIDAQAB
+-----END PUBLIC KEY-----
+"""

api/organisations/management/commands/create_licensing_private_key.py

@@ -0,0 +1,10 @@
+from typing import Any
+
+from django.core.management import BaseCommand
+
+from organisations.subscriptions.licensing.helpers import create_private_key
+
+
+class Command(BaseCommand):
+    def handle(self, *args: Any, **options: Any) -> None:
+        print(create_private_key())

api/organisations/management/commands/create_licensing_public_key.py

@@ -0,0 +1,10 @@
+from typing import Any
+
+from django.core.management import BaseCommand
+
+from organisations.subscriptions.licensing.helpers import create_public_key
+
+
+class Command(BaseCommand):
+    def handle(self, *args: Any, **options: Any) -> None:
+        print(create_public_key())

api/organisations/models.py

@@ -415,16 +415,16 @@ def _get_subscription_metadata_for_chargebee(self) -> ChargebeeObjMetadata:
         return cb_metadata
 
     def _get_subscription_metadata_for_self_hosted(self) -> BaseSubscriptionMetadata:
-        if not is_enterprise():
-            return FREE_PLAN_SUBSCRIPTION_METADATA
+        if is_enterprise() and hasattr(self.organisation, "licence"):
+            licence_information = self.organisation.licence.get_licence_information()
+            return BaseSubscriptionMetadata(
+                seats=licence_information.num_seats,
+                projects=licence_information.num_projects,
+                audit_log_visibility_days=None,
+                feature_history_visibility_days=None,
+            )
 
-        return BaseSubscriptionMetadata(
-            seats=self.max_seats,
-            api_calls=self.max_api_calls,
-            projects=None,
-            audit_log_visibility_days=None,
-            feature_history_visibility_days=None,
-        )
+        return FREE_PLAN_SUBSCRIPTION_METADATA
 
     def add_single_seat(self):
         if not self.can_auto_upgrade_seats:

api/organisations/subscriptions/licensing/__init__.py

@@ -0,0 +1 @@
+# TODO: split this into a private package?

api/organisations/subscriptions/licensing/helpers.py

@@ -0,0 +1,91 @@
+import base64
+import logging
+
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import padding, rsa
+from django.conf import settings
+
+logger: logging.Logger = logging.getLogger(name=__name__)
+
+
+def sign_licence(licence: str) -> str:
+    message = licence.encode("utf-8")
+
+    # Load the private key from PEM
+    private_key = serialization.load_pem_private_key(
+        settings.SUBSCRIPTION_LICENCE_PRIVATE_KEY.encode("utf-8"), password=None
+    )
+
+    # Sign the message using the private key
+    signature = private_key.sign(
+        message,
+        padding.PSS(
+            mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH
+        ),
+        hashes.SHA256(),
+    )
+
+    return base64.b64encode(signature).decode("utf-8")
+
+
+def verify_signature(licence: str, licence_signature: str) -> bool:
+    signature = base64.b64decode(licence_signature)
+    public_key = serialization.load_pem_public_key(
+        settings.SUBSCRIPTION_LICENCE_PUBLIC_KEY.encode("utf-8")
+    )
+
+    try:
+        public_key.verify(
+            signature,
+            licence.encode("utf-8"),
+            padding.PSS(
+                mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH
+            ),
+            hashes.SHA256(),
+        )
+    except Exception:
+        logger.error("Licence signature failed", exc_info=True)
+        return False
+
+    return True
+
+
+def create_public_key() -> str:
+    """
+    Creates a public key from the private key that's set in settings.
+    """
+
+    # Load the private key from the UTF-8 PEM string.
+    private_key = serialization.load_pem_private_key(
+        settings.SUBSCRIPTION_LICENCE_PRIVATE_KEY.encode("utf-8"),
+        password=None,
+        backend=default_backend(),
+    )
+
+    # Extract the public key from the private key.
+    public_key = private_key.public_key()
+
+    # Encode the public key to PEM format
+    public_key_pem = public_key.public_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+
+    return public_key_pem.decode("utf-8")
+
+
+def create_private_key() -> str:
+    private_key = rsa.generate_private_key(
+        public_exponent=65537, key_size=2048, backend=default_backend()
+    )
+
+    # Convert the private key to PEM format as a byte string
+    private_key_pem = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
+    )
+
+    # Print the PEM-encoded private key to standard output
+    return private_key_pem.decode("utf-8")

api/organisations/subscriptions/licensing/licensing.py

@@ -0,0 +1,18 @@
+import typing
+from datetime import datetime
+
+from pydantic import BaseModel
+
+
+class LicenceInformation(BaseModel):
+    organisation_name: str
+    plan_id: str
+
+    department_name: typing.Optional[str] = None
+    expiry_date: typing.Optional[datetime] = None
+
+    num_seats: int
+    num_projects: int  # TODO: what about Flagsmith on Flagsmith project?
+    num_api_calls: typing.Optional[int] = (
+        None  # required to support private cloud installs
+    )

api/organisations/subscriptions/licensing/migrations/0001_initial.py

@@ -0,0 +1,26 @@
+# Generated by Django 3.2.24 on 2024-03-15 15:52
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('organisations', '0052_create_hubspot_organisation'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='OrganisationLicence',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('updated_at', models.DateTimeField(auto_now=True)),
+                ('content', models.TextField(blank=True)),
+                ('organisation', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, to='organisations.organisation')),
+            ],
+        ),
+    ]

api/organisations/subscriptions/licensing/migrations/0002_alter_organisationlicence_organisation.py

@@ -0,0 +1,24 @@
+# Generated by Django 4.2.16 on 2024-10-30 15:40
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("organisations", "0058_update_audit_and_history_limits_in_sub_cache"),
+        ("licensing", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="organisationlicence",
+            name="organisation",
+            field=models.OneToOneField(
+                on_delete=django.db.models.deletion.CASCADE,
+                related_name="licence",
+                to="organisations.organisation",
+            ),
+        ),
+    ]

api/organisations/subscriptions/licensing/models.py

@@ -0,0 +1,17 @@
+from django.db import models
+
+from organisations.subscriptions.licensing.licensing import LicenceInformation
+
+
+class OrganisationLicence(models.Model):
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    organisation = models.OneToOneField(
+        "organisations.Organisation", related_name="licence", on_delete=models.CASCADE
+    )
+
+    content = models.TextField(blank=True)
+
+    def get_licence_information(self) -> LicenceInformation:
+        return LicenceInformation.parse_raw(self.content)

api/organisations/subscriptions/licensing/views.py

@@ -0,0 +1,30 @@
+from rest_framework import serializers
+from rest_framework.decorators import api_view
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from organisations.subscriptions.licensing.helpers import verify_signature
+from organisations.subscriptions.licensing.models import OrganisationLicence
+
+
+@api_view(http_method_names=["PUT"])
+def create_or_update_licence(
+    request: Request, organisation_id: int, **kwargs
+) -> Response:
+    if "licence" not in request.FILES:
+        raise serializers.ValidationError("No licence file provided.")
+
+    if "licence_signature" not in request.FILES:
+        raise serializers.ValidationError("No licence signature file provided.")
+
+    licence = request.FILES["licence"].read().decode("utf-8")
+    licence_signature = request.FILES["licence_signature"].read().decode("utf-8")
+
+    if verify_signature(licence, licence_signature):
+        OrganisationLicence.objects.update_or_create(
+            organisation_id=organisation_id,
+            defaults={"content": licence},
+        )
+    else:
+        raise serializers.ValidationError("Signature failed for licence.")
+    return Response(200)

api/organisations/subscriptions/metadata.py

@@ -9,13 +9,13 @@ class BaseSubscriptionMetadata:
     def __init__(
         self,
         seats: int = 0,
-        api_calls: int = 0,
-        projects: typing.Optional[int] = None,
-        chargebee_email: str = None,
+        api_calls: None | int = None,
+        projects: None | int = None,
+        chargebee_email: None | str = None,
         audit_log_visibility_days: int | None = 0,
         feature_history_visibility_days: int | None = DEFAULT_VERSION_LIMIT_DAYS,
         **kwargs,  # allows for extra unknown attrs from CB json metadata
-    ):
+    ) -> None:
         self.seats = seats
         self.api_calls = api_calls
         self.projects = projects

api/organisations/urls.py

@@ -35,6 +35,7 @@
     UserOrganisationPermissionViewSet,
     UserPermissionGroupOrganisationPermissionViewSet,
 )
+from .subscriptions.licensing.views import create_or_update_licence
 
 router = routers.DefaultRouter()
 router.register(r"", views.OrganisationViewSet, basename="organisation")
@@ -152,6 +153,11 @@
         OrganisationAPIUsageNotificationView.as_view(),
         name="organisation-api-usage-notification",
     ),
+    path(
+        "<int:organisation_id>/licence",
+        create_or_update_licence,
+        name="create-or-update-licence",
+    ),
 ]
 
 if settings.IS_RBAC_INSTALLED: