feat: Legacy file validation.

kirovy/constants/__init__.py

@@ -90,12 +90,18 @@ def is_messiah(cls, user_group: str) -> bool:
 
 
 class MigrationUser:
-    ID = -1
+    CNCNET_ID = -1
     USERNAME = "MobileConstructionVehicle_Migrator"
     GROUP = CncnetUserGroup.USER
 
 
-class GameSlugs(enum.StrEnum):
+class LegacyUploadUser:
+    CNCNET_ID = -2
+    USERNAME = "Spy_ShapeShifting_LegacyUploader"
+    GROUP = CncnetUserGroup.USER
+
+
+class GameSlugs(str, enum.Enum):
     """The slugs for each game / total conversion mod.
 
     These **must** be unique. They are in constants because we need them to determine which parser to use

kirovy/constants/api_codes.py

@@ -3,8 +3,19 @@
 
 class UploadApiCodes(enum.StrEnum):
     GAME_SLUG_DOES_NOT_EXIST = "game-slug-does-not-exist"
+    GAME_DOES_NOT_EXIST = "game-does-not-exist"
     MISSING_GAME_SLUG = "missing-game-slug"
     FILE_TO_LARGE = "file-too-large"
     EMPTY_UPLOAD = "where-file"
     DUPLICATE_MAP = "duplicate-map"
     FILE_EXTENSION_NOT_SUPPORTED = "file-extension-not-supported"
+
+
+class LegacyUploadApiCodes(enum.StrEnum):
+    NOT_A_VALID_ZIP_FILE = "invalid-zipfile"
+    BAD_ZIP_STRUCTURE = "invalid-zip-structure"
+    MAP_TOO_LARGE = "map-file-too-large"
+    NO_VALID_MAP_FILE = "no-valid-map-file"
+    HASH_MISMATCH = "file-hash-does-not-match-zip-name"
+    INVALID_FILE_TYPE = "invalid-file-type-in-zip"
+    GAME_NOT_SUPPORTED = "game-not-supported"

kirovy/models/cnc_user.py

@@ -15,7 +15,8 @@ class CncUserManager(models.Manager):
     use_in_migrations = True
 
     _SYSTEM_CNCNET_IDS = {
-        constants.MigrationUser.ID,
+        constants.MigrationUser.CNCNET_ID,
+        constants.LegacyUploadUser.CNCNET_ID,
     }
 
     def find_by_cncnet_id(self, cncnet_id: int) -> t.Tuple["CncUser"]:
@@ -27,10 +28,10 @@ def get_or_create_migration_user(self) -> "CncUser":
         :return:
             The user for running migrations.
         """
-        mcv = self.find_by_cncnet_id(constants.MigrationUser.ID)
+        mcv = self.find_by_cncnet_id(constants.MigrationUser.CNCNET_ID)
         if not mcv:
             mcv = CncUser(
-                cncnet_id=constants.MigrationUser.ID,
+                cncnet_id=constants.MigrationUser.CNCNET_ID,
                 username=constants.MigrationUser.USERNAME,
                 group=constants.MigrationUser.GROUP,
             )
@@ -39,6 +40,30 @@ def get_or_create_migration_user(self) -> "CncUser":
 
         return mcv
 
+    def get_or_create_legacy_upload_user(self) -> "CncUser":
+        """Gets or creates a system-user to represent anonymous uploads from a CnCNet client.
+
+        .. warning::
+
+            This should **only** be used for the legacy upload URLs for clients
+            that CnCNet doesn't have the source for.
+
+        :return:
+            User for legacy uploads.
+        """
+        # If we copy and paste this again then it should be DRY'd up.
+        spy = self.find_by_cncnet_id(constants.LegacyUploadUser.CNCNET_ID)
+        if not spy:
+            spy = CncUser(
+                cncnet_id=constants.LegacyUploadUser.CNCNET_ID,
+                username=constants.LegacyUploadUser.USERNAME,
+                group=constants.LegacyUploadUser.GROUP,
+            )
+            spy.save()
+            spy.refresh_from_db()
+
+        return spy
+
     def get_queryset(self) -> models.QuerySet:
         """Makes ``CncUser.object.all()`` filter out the system users by default.
 
@@ -61,9 +86,7 @@ class CncUser(AbstractBaseUser):
         help_text=_("The user ID from the CNCNet ladder API."),
     )
 
-    username = models.CharField(
-        null=True, help_text=_("The name from the CNCNet ladder API."), blank=False
-    )
+    username = models.CharField(null=True, help_text=_("The name from the CNCNet ladder API."), blank=False)
     """:attr: The username for debugging purposes. Don't rely on this field for much else."""
 
     verified_map_uploader = models.BooleanField(null=False, default=False)
@@ -77,23 +100,15 @@ class CncUser(AbstractBaseUser):
         blank=False,
     )
 
-    is_banned = models.BooleanField(
-        default=False, help_text="If true, user was banned for some reason."
-    )
-    ban_reason = models.CharField(
-        default=None, null=True, help_text="If banned, the reason the user was banned."
-    )
-    ban_date = models.DateTimeField(
-        default=None, null=True, help_text="If banned, when the user was banned."
-    )
+    is_banned = models.BooleanField(default=False, help_text="If true, user was banned for some reason.")
+    ban_reason = models.CharField(default=None, null=True, help_text="If banned, the reason the user was banned.")
+    ban_date = models.DateTimeField(default=None, null=True, help_text="If banned, when the user was banned.")
     ban_expires = models.DateTimeField(
         default=None,
         null=True,
         help_text="If banned, when the ban expires, if temporary.",
     )
-    ban_count = models.IntegerField(
-        default=0, help_text="How many times this user has been banned."
-    )
+    ban_count = models.IntegerField(default=0, help_text="How many times this user has been banned.")
 
     USERNAME_FIELD = "cncnet_id"
     """:attr:
@@ -113,9 +128,7 @@ def can_upload(self) -> bool:
         :return:
             True if user can upload maps / mixes / big, or edit their existing uploads.
         """
-        self.refresh_from_db(
-            fields=["verified_map_uploader", "verified_email", "is_banned"]
-        )
+        self.refresh_from_db(fields=["verified_map_uploader", "verified_email", "is_banned"])
         can_upload = self.verified_map_uploader or self.verified_email or self.is_staff
         return can_upload and not self.is_banned
 
@@ -142,9 +155,7 @@ def create_or_update_from_cncnet(user_dto: CncnetUserInfo) -> "CncUser":
         :return:
             The user object in Kirovy's database, updated with the data from CnCNet.
         """
-        kirovy_user: t.Optional[CncUser] = CncUser.objects.filter(
-            cncnet_id=user_dto.id
-        ).first()
+        kirovy_user: t.Optional[CncUser] = CncUser.objects.filter(cncnet_id=user_dto.id).first()
         if not kirovy_user:
             kirovy_user = CncUser.objects.create(
                 cncnet_id=user_dto.id,

kirovy/models/file_base.py

@@ -89,9 +89,12 @@ def validate_file_extension(self, file_extension: game_models.CncFileExtension)
     def save(self, *args, **kwargs):
         self.validate_file_extension(self.file_extension)
 
-        self.hash_md5 = file_utils.hash_file_md5(self.file)
-        self.hash_sha512 = file_utils.hash_file_sha512(self.file)
-        self.hash_sha1 = file_utils.hash_file_sha1(self.file)
+        if not self.hash_md5:
+            self.hash_md5 = file_utils.hash_file_md5(self.file)
+        if not self.hash_sha512:
+            self.hash_sha512 = file_utils.hash_file_sha512(self.file)
+        if not self.hash_sha1:
+            self.hash_sha1 = file_utils.hash_file_sha1(self.file)
         super().save(*args, **kwargs)
 
     @staticmethod

kirovy/services/legacy_upload/__init__.py

@@ -0,0 +1,23 @@
+from kirovy.constants import GameSlugs
+from kirovy.constants.api_codes import LegacyUploadApiCodes
+from kirovy.exceptions import view_exceptions
+from kirovy.services.legacy_upload import westwood, dune_2000
+from kirovy.services.legacy_upload.base import LegacyMapServiceBase
+from kirovy import typing as t
+
+
+_GAME_LEGACY_SERVICE_MAP: t.Dict[str, t.Type[LegacyMapServiceBase]] = {
+    GameSlugs.yuris_revenge.value: westwood.YurisRevengeLegacyMapService,
+    GameSlugs.dune_2000.value: dune_2000.Dune2000LegacyMapService,
+}
+
+
+def get_legacy_service_for_slug(game_slug: str) -> t.Type[LegacyMapServiceBase]:
+    if service := _GAME_LEGACY_SERVICE_MAP.get(game_slug):
+        return service
+
+    raise view_exceptions.KirovyValidationError(
+        "Game not supported on legacy endpoint",
+        code=LegacyUploadApiCodes.GAME_NOT_SUPPORTED,
+        additional={"supported": _GAME_LEGACY_SERVICE_MAP.keys()},
+    )