Limit the number of tags that can be sent (#1133)

core/schemas/tag.py

@@ -12,6 +12,7 @@
 DEFAULT_EXPIRATION = datetime.timedelta(days=30)  # Completely arbitrary
 
 MAX_TAG_LENGTH = 50
+MAX_TAGS_REQUEST = 50
 
 
 def future():

core/web/apiv2/entities.py

@@ -1,16 +1,17 @@
 from fastapi import APIRouter, HTTPException
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, conlist
 
 from core.schemas import graph
 from core.schemas.entity import Entity, EntityType, EntityTypes
+from core.schemas.tag import MAX_TAGS_REQUEST
 
 
 # Request schemas
 class NewEntityRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
     entity: EntityTypes = Field(discriminator="type")
-    tags: list[str] = []
+    tags: conlist(str, max_length=MAX_TAGS_REQUEST) = []
 
 
 class PatchEntityRequest(BaseModel):
@@ -41,7 +42,7 @@ class EntityTagRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
     ids: list[str]
-    tags: list[str]
+    tags: conlist(str, max_length=MAX_TAGS_REQUEST) = []
     strict: bool = False
 
 

core/web/apiv2/graph.py

@@ -3,12 +3,13 @@
 from typing import Any
 
 from fastapi import APIRouter, HTTPException
-from pydantic import BaseModel, ConfigDict, ValidationInfo, model_validator
+from pydantic import BaseModel, ConfigDict, ValidationInfo, conlist, model_validator
 from pydantic.functional_validators import field_validator
 
 from core.schemas import dfiq, entity, graph, indicator, observable, tag
 from core.schemas.graph import GraphFilter
 from core.schemas.observable import ObservableType
+from core.schemas.tag import MAX_TAGS_REQUEST
 
 GRAPH_TYPE_MAPPINGS = {}  # type: dict[str, Type[entity.Entity] | Type[observable.Observable] | Type[indicator.Indicator]]
 GRAPH_TYPE_MAPPINGS.update(observable.TYPE_MAPPING)
@@ -213,7 +214,7 @@ async def delete(relationship_id: str) -> None:
 
 class AnalysisRequest(BaseModel):
     observables: list[str]
-    add_tags: list[str] = []
+    add_tags: conlist(str, max_length=MAX_TAGS_REQUEST) = []
     regex_match: bool = False
     add_type: observable.ObservableType | None = None
     fetch_neighbors: bool = True

core/web/apiv2/indicators.py

@@ -1,5 +1,5 @@
 from fastapi import APIRouter, HTTPException
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, conlist
 
 from core.schemas import graph
 from core.schemas.indicator import (
@@ -8,6 +8,7 @@
     IndicatorType,
     IndicatorTypes,
 )
+from core.schemas.tag import MAX_TAGS_REQUEST
 
 
 # Request schemas
@@ -45,7 +46,7 @@ class IndicatorTagRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
     ids: list[str]
-    tags: list[str]
+    tags: conlist(str, max_length=MAX_TAGS_REQUEST) = []
     strict: bool = False
 
 

core/web/apiv2/observables.py

@@ -1,11 +1,11 @@
-from typing import Iterable
+from typing import Annotated, Iterable
 
 from fastapi import APIRouter, HTTPException
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, conlist, field_validator
 
 from core.schemas import graph
 from core.schemas.observable import TYPE_MAPPING, Observable, ObservableType
-from core.schemas.tag import MAX_TAG_LENGTH
+from core.schemas.tag import MAX_TAG_LENGTH, MAX_TAGS_REQUEST
 
 ObservableTypes = ()
 
@@ -20,7 +20,7 @@
 
 
 class TagRequestMixin(BaseModel):
-    tags: list[str] = []
+    tags: conlist(str, max_length=MAX_TAGS_REQUEST) = []
 
     @field_validator("tags")
     @classmethod

tests/apiv2/observables.py

@@ -243,6 +243,20 @@ def test_create_observable_toolong_tag(self):
             data,
         )
 
+    def test_create_observable_toomany_tags(self):
+        many_tags = [str(i) for i in range(200)]
+        response = client.post(
+            "/api/v2/observables/",
+            json={"value": "toto.com", "type": "hostname", "tags": many_tags},
+        )
+        data = response.json()
+        self.assertEqual(response.status_code, 422, data)
+        self.assertEqual(
+            data["detail"][0]["msg"],
+            "List should have at most 50 items after validation, not 200",
+            data,
+        )
+
     def test_create_extended_observable(self):
         response = client.post(
             "/api/v2/observables/extended",