From a534c88ca0e2e97231dd5d5ed4019f4f9d32c32f Mon Sep 17 00:00:00 2001 From: Rafia minhaj Date: Thu, 2 Jul 2026 01:00:12 +0530 Subject: [PATCH] feat(validation): introduce string length constraints on policy, profile, and workflow endpoints --- backend/secuscan/models.py | 8 +-- backend/secuscan/routes.py | 58 ++++++++++++++- testing/backend/test_string_length_limits.py | 75 ++++++++++++++++++++ 3 files changed, 135 insertions(+), 6 deletions(-) create mode 100644 testing/backend/test_string_length_limits.py diff --git a/backend/secuscan/models.py b/backend/secuscan/models.py index e04f2f12d..03d8e2a86 100644 --- a/backend/secuscan/models.py +++ b/backend/secuscan/models.py @@ -310,19 +310,19 @@ class NotificationDeliveryStatus(str, Enum): class NotificationRuleCreate(BaseModel): """Request payload for creating or updating a notification rule.""" - name: str + name: str = Field(..., max_length=255) severity_threshold: NotificationSeverityThreshold channel_type: NotificationChannelType - target_url_or_email: str + target_url_or_email: str = Field(..., max_length=2000) is_active: bool = True class NotificationRuleUpdate(BaseModel): """Partial update payload for a notification rule.""" - name: Optional[str] = None + name: Optional[str] = Field(default=None, max_length=255) severity_threshold: Optional[NotificationSeverityThreshold] = None channel_type: Optional[NotificationChannelType] = None - target_url_or_email: Optional[str] = None + target_url_or_email: Optional[str] = Field(default=None, max_length=2000) is_active: Optional[bool] = None diff --git a/backend/secuscan/routes.py b/backend/secuscan/routes.py index b487a1bc4..6f6dd2a37 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -1564,11 +1564,36 @@ async def list_target_policies(owner: str = Depends(get_current_owner)): return {"items": deserialize_resource_rows(rows), "total": len(rows)} +def _validate_lengths( + name: Optional[str] = None, + description: Optional[str] = None, + notes: Optional[str] = None, + resource_type: str = "Resource", +): + if name is not None and len(str(name).strip()) > 255: + raise HTTPException( + status_code=400, + detail=f"{resource_type} name exceeds maximum length of 255 characters", + ) + if description is not None and len(str(description).strip()) > 2000: + raise HTTPException( + status_code=400, + detail=f"{resource_type} description exceeds maximum length of 2000 characters", + ) + if notes is not None and len(str(notes).strip()) > 2000: + raise HTTPException( + status_code=400, + detail=f"{resource_type} notes exceeds maximum length of 2000 characters", + ) + + @router.post("/target-policies", dependencies=[Depends(admin_limiter)]) async def create_target_policy(payload: Dict[str, Any], owner: str = Depends(get_current_owner)): name = str(payload.get("name", "")).strip() if not name: raise HTTPException(status_code=400, detail="Target policy name is required") + description = str(payload.get("description", "")).strip() + _validate_lengths(name=name, description=description, resource_type="Target policy") allowed = payload.get("allowed_targets") if allowed is not None and not isinstance(allowed, list): @@ -1588,7 +1613,7 @@ async def create_target_policy(payload: Dict[str, Any], owner: str = Depends(get policy_id, owner, name, - str(payload.get("description", "")).strip() or None, + description or None, 1 if payload.get("allow_public_targets") else 0, 1 if payload.get("allow_exploit_validation") else 0, 1 if payload.get("allow_authenticated_scan") else 0, @@ -1607,6 +1632,14 @@ async def update_target_policy(policy_id: str, payload: Dict[str, Any], owner: s row = await db.fetchone("SELECT id FROM target_policies WHERE id = ? AND owner_id = ?", (policy_id, owner)) if not row: raise HTTPException(status_code=404, detail="Target policy not found") + + if "name" in payload or "description" in payload: + _validate_lengths( + name=payload.get("name"), + description=payload.get("description"), + resource_type="Target policy", + ) + updates: List[str] = [] params: List[Any] = [] for key in ("name", "description", "default_validation_mode"): @@ -1652,6 +1685,8 @@ async def create_credential_profile(payload: Dict[str, Any], owner: str = Depend name = str(payload.get("name", "")).strip() if not name: raise HTTPException(status_code=400, detail="Credential profile name is required") + _validate_lengths(name=name, resource_type="Credential profile") + profile_id = str(uuid.uuid4()) db = await get_db() await db.execute( @@ -1681,6 +1716,10 @@ async def update_credential_profile(profile_id: str, payload: Dict[str, Any], ow row = await db.fetchone("SELECT id FROM credential_profiles WHERE id = ? AND owner_id = ?", (profile_id, owner)) if not row: raise HTTPException(status_code=404, detail="Credential profile not found") + + if "name" in payload: + _validate_lengths(name=payload.get("name"), resource_type="Credential profile") + updates: List[str] = [] params: List[Any] = [] for key in ("name", "username_secret_name", "password_secret_name"): @@ -1722,6 +1761,9 @@ async def create_session_profile(payload: Dict[str, Any], owner: str = Depends(g name = str(payload.get("name", "")).strip() if not name: raise HTTPException(status_code=400, detail="Session profile name is required") + notes = str(payload.get("notes", "")).strip() + _validate_lengths(name=name, notes=notes, resource_type="Session profile") + profile_id = str(uuid.uuid4()) db = await get_db() await db.execute( @@ -1736,7 +1778,7 @@ async def create_session_profile(payload: Dict[str, Any], owner: str = Depends(g name, payload.get("cookie_secret_name"), _json_payload(payload.get("extra_headers"), "{}"), - str(payload.get("notes", "")).strip() or None, + notes or None, ), ) row = await db.fetchone("SELECT * FROM session_profiles WHERE id = ?", (profile_id,)) @@ -1749,6 +1791,14 @@ async def update_session_profile(profile_id: str, payload: Dict[str, Any], owner row = await db.fetchone("SELECT id FROM session_profiles WHERE id = ? AND owner_id = ?", (profile_id, owner)) if not row: raise HTTPException(status_code=404, detail="Session profile not found") + + if "name" in payload or "notes" in payload: + _validate_lengths( + name=payload.get("name"), + notes=payload.get("notes"), + resource_type="Session profile", + ) + updates: List[str] = [] params: List[Any] = [] for key in ("name", "cookie_secret_name", "notes"): @@ -1813,6 +1863,7 @@ async def create_workflow(payload: Dict[str, Any], owner: str = Depends(get_curr name = str(payload.get("name", "")).strip() if not name: raise HTTPException(status_code=400, detail="Workflow name is required") + _validate_lengths(name=name, resource_type="Workflow") steps = _parse_workflow_steps(payload.get("steps", [])) if not steps: @@ -1997,6 +2048,9 @@ async def update_workflow(workflow_id: str, payload: Dict[str, Any], owner: str db = await get_db() row = await _verify_workflow_owner(db, workflow_id, owner) + if "name" in payload: + _validate_lengths(name=payload.get("name"), resource_type="Workflow") + old_enabled = bool(row["enabled"]) new_enabled = old_enabled enabled_changed = False diff --git a/testing/backend/test_string_length_limits.py b/testing/backend/test_string_length_limits.py new file mode 100644 index 000000000..d51af440a --- /dev/null +++ b/testing/backend/test_string_length_limits.py @@ -0,0 +1,75 @@ +import pytest + +def test_target_policy_length_constraints(test_client): + # Name > 255 characters + long_name = "a" * 256 + response = test_client.post("/api/v1/target-policies", json={"name": long_name}) + assert response.status_code == 400 + assert "name exceeds maximum length of 255 characters" in response.json()["detail"] + + # Description > 2000 characters + long_desc = "a" * 2001 + response = test_client.post("/api/v1/target-policies", json={"name": "Valid Name", "description": long_desc}) + assert response.status_code == 400 + assert "description exceeds maximum length of 2000 characters" in response.json()["detail"] + + +def test_credential_profile_length_constraints(test_client): + # Name > 255 characters + long_name = "a" * 256 + response = test_client.post("/api/v1/credential-profiles", json={"name": long_name}) + assert response.status_code == 400 + assert "name exceeds maximum length of 255 characters" in response.json()["detail"] + + +def test_session_profile_length_constraints(test_client): + # Name > 255 characters + long_name = "a" * 256 + response = test_client.post("/api/v1/session-profiles", json={"name": long_name}) + assert response.status_code == 400 + assert "name exceeds maximum length of 255 characters" in response.json()["detail"] + + # Notes > 2000 characters + long_notes = "a" * 2001 + response = test_client.post("/api/v1/session-profiles", json={"name": "Valid Name", "notes": long_notes}) + assert response.status_code == 400 + assert "notes exceeds maximum length of 2000 characters" in response.json()["detail"] + + +def test_workflow_length_constraints(test_client): + # Name > 255 characters + long_name = "a" * 256 + response = test_client.post( + "/api/v1/workflows", + json={"name": long_name, "steps": [{"plugin_id": "port_scan", "inputs": {}}]} + ) + assert response.status_code == 400 + assert "name exceeds maximum length of 255 characters" in response.json()["detail"] + + +def test_notification_rule_length_constraints(test_client): + # Pydantic validation: Name > 255 characters + long_name = "a" * 256 + response = test_client.post( + "/api/v1/notifications/rules", + json={ + "name": long_name, + "severity_threshold": "high", + "channel_type": "email", + "target_url_or_email": "test@example.com" + } + ) + assert response.status_code == 422 + + # Pydantic validation: Target > 2000 characters + long_target = "a" * 2001 + "@example.com" + response = test_client.post( + "/api/v1/notifications/rules", + json={ + "name": "Valid Rule", + "severity_threshold": "high", + "channel_type": "email", + "target_url_or_email": long_target + } + ) + assert response.status_code == 422