diff --git a/services/flask-backend/app/__init__.py b/services/flask-backend/app/__init__.py index fa711cdd..8878f7ba 100644 --- a/services/flask-backend/app/__init__.py +++ b/services/flask-backend/app/__init__.py @@ -262,6 +262,13 @@ def _register_blueprints(app: Quart) -> None: app.register_blueprint(tenants_bp, url_prefix="/api/v1") + # URL shortener management (Phase 3) + from .urls import urls_bp + from .collections import collections_bp + + app.register_blueprint(urls_bp, url_prefix="/api/v1") + app.register_blueprint(collections_bp, url_prefix="/api/v1") + # OAuth2 token endpoints (Phase 2) from .oauth import oauth_bp diff --git a/services/flask-backend/app/collections.py b/services/flask-backend/app/collections.py new file mode 100644 index 00000000..29e8dc87 --- /dev/null +++ b/services/flask-backend/app/collections.py @@ -0,0 +1,446 @@ +""" +Collection Management API Endpoints. + +Implements collection CRUD operations with scope-based permissions. +Supports nested collections (folders) within the same tenant. +All operations are tenant-scoped. +""" + +from __future__ import annotations + +from pydantic import ValidationError +from quart import Blueprint, g, jsonify, request +from werkzeug.exceptions import BadRequest, Forbidden, NotFound + +from .auth import auth_required +from .models import get_db +from .rbac import get_user_scopes, require_scope +from .schemas.collections import ( + CollectionCreateRequest, + CollectionDetailResponse, + CollectionListResponse, + CollectionResponse, + CollectionUpdateRequest, +) + +collections_bp = Blueprint("collections", __name__) + + +def _is_collection_member_or_admin( + db, user_id: int, collection_id: int, tenant_id: int +) -> bool: + """ + Check if user is the creator of the collection or is a tenant admin. + + Args: + db: Database instance + user_id: Current user ID + collection_id: Collection ID to check + tenant_id: Tenant ID for the collection + + Returns: + True if user created the collection or is tenant admin + """ + collection = db(db.collections.id == collection_id).select().first() + if not collection: + return False + + # Creator can edit own collections + if collection.created_by == user_id: + return True + + # Tenant admin can edit any collection in their tenant + from .rbac import _has_tenant_admin_scope + + return _has_tenant_admin_scope(db, user_id, tenant_id) + + +def _validate_no_cycles(db, collection_id: int, parent_id: int) -> None: + """ + Validate that setting parent_id doesn't create a cycle. + + Checks if collection_id is in the ancestors of parent_id. + If collection_id appears in the parent chain of parent_id, it means + collection_id is already an ancestor of parent_id, so making parent_id + the parent of collection_id would create a cycle. + + Args: + db: Database instance + collection_id: Collection being updated + parent_id: Proposed parent collection ID + + Raises: + BadRequest: If a cycle would be created + """ + if parent_id == collection_id: + raise BadRequest("A collection cannot be its own parent") + + # Walk up the parent chain of parent_id to see if we encounter collection_id + # If we do, it means collection_id is an ancestor of parent_id, creating a cycle + visited = set() + current_id = parent_id + + while current_id is not None: + if current_id == collection_id: + raise BadRequest("Cannot set parent: would create a cycle") + + if current_id in visited: + # Sanity check - should not happen in a well-formed tree + break + + visited.add(current_id) + + parent = db(db.collections.id == current_id).select().first() + if not parent: + break + current_id = parent.parent_id + + +@collections_bp.route("/collections", methods=["POST"]) +@auth_required +@require_scope("collections:write", "collections:admin") +async def create_collection(): + """ + Create a new collection. + + Supports nested collections via parent_id. + + Requires: collections:write or collections:admin scope + """ + data = await request.get_json() + if not data: + raise BadRequest("Request body required") + + try: + req = CollectionCreateRequest(**data) + except BadRequest as e: + # Validation error from field validators + error_msg = str(e).replace("400 Bad Request: ", "") + return jsonify({"error": error_msg}), 400 + except ValidationError as e: + return jsonify({"error": e.errors()[0].get("msg", "Validation error")}), 400 + + db = get_db() + user_id = g.current_user["id"] + tenant_id = g.current_user.get("tenant_id") + + if not tenant_id: + default_tenant = db(db.tenants.is_default == True).select().first() # noqa + if not default_tenant: + raise BadRequest("No tenant found for user") + tenant_id = default_tenant.id + + # Verify parent exists and belongs to same tenant if specified + try: + if req.parent_id: + parent = db( + (db.collections.id == req.parent_id) + & (db.collections.tenant_id == tenant_id) + ).select().first() + if not parent: + raise BadRequest("Parent collection not found or invalid") + + # Verify team if specified + if req.team_id: + team = db(db.teams.id == req.team_id).select().first() + if not team or team.tenant_id != tenant_id: + raise BadRequest("Invalid team ID") + except BadRequest as e: + error_msg = str(e).replace("400 Bad Request: ", "") + return jsonify({"error": error_msg}), 400 + + # Check name uniqueness within tenant + parent scope + existing = db( + (db.collections.tenant_id == tenant_id) + & (db.collections.name == req.name) + & (db.collections.parent_id == req.parent_id) + ).select().first() + if existing: + return jsonify({"error": "Collection name already exists in this location"}), 409 + + collection_id = db.collections.insert( + tenant_id=tenant_id, + team_id=req.team_id, + name=req.name, + description=req.description, + parent_id=req.parent_id, + created_by=user_id, + ) + db.commit() + + collection = db(db.collections.id == collection_id).select().first() + response_data = CollectionResponse( + id=collection.id, + tenant_id=collection.tenant_id, + team_id=collection.team_id, + name=collection.name, + description=collection.description, + parent_id=collection.parent_id, + created_by=collection.created_by, + created_at=collection.created_at, + ) + + return jsonify({"data": response_data.model_dump(mode="json")}), 201 + + +@collections_bp.route("/collections", methods=["GET"]) +@auth_required +@require_scope("collections:read") +async def list_collections(): + """ + List collections with pagination and optional filtering. + + Query params: + - limit: Items per page (default 20, max 100) + - offset: Pagination offset (default 0) + - parent_id: Filter by parent (for listing children of a collection) + - flat: If 'true', return flat list; if 'false', return tree structure (default flat) + + Requires: collections:read scope + """ + db = get_db() + tenant_id = g.current_user.get("tenant_id") + + if not tenant_id: + default_tenant = db(db.tenants.is_default == True).select().first() # noqa + if not default_tenant: + raise BadRequest("No tenant found for user") + tenant_id = default_tenant.id + + # Pagination + try: + limit = int(request.args.get("limit", 20)) + offset = int(request.args.get("offset", 0)) + limit = min(limit, 100) + except ValueError: + raise BadRequest("Invalid limit/offset") + + # Build query + query = db.collections.tenant_id == tenant_id + + # Optional parent filter + parent_id_param = request.args.get("parent_id") + if parent_id_param: + if parent_id_param.lower() == "none": + query &= db.collections.parent_id == None # noqa + else: + try: + parent_id = int(parent_id_param) + query &= db.collections.parent_id == parent_id + except ValueError: + raise BadRequest("Invalid parent_id") + + # Count total + total = db(query).count() + + # Fetch with pagination + rows = ( + db(query) + .select(orderby=db.collections.name, limitby=(offset, offset + limit)) + .as_list() + ) + + data = [ + CollectionResponse( + id=r["id"], + tenant_id=r["tenant_id"], + team_id=r["team_id"], + name=r["name"], + description=r["description"], + parent_id=r["parent_id"], + created_by=r["created_by"], + created_at=r["created_at"], + ) + for r in rows + ] + + response = CollectionListResponse( + data=data, + total=total, + limit=limit, + offset=offset, + ) + return jsonify(response.model_dump(mode="json")), 200 + + +@collections_bp.route("/collections/", methods=["GET"]) +@auth_required +@require_scope("collections:read") +async def get_collection(collection_id: int): + """ + Get collection details by ID. + + Requires: collections:read scope (tenant-scoped) + """ + db = get_db() + tenant_id = g.current_user.get("tenant_id") + + if not tenant_id: + default_tenant = db(db.tenants.is_default == True).select().first() # noqa + if not default_tenant: + raise BadRequest("No tenant found for user") + tenant_id = default_tenant.id + + collection = ( + db( + (db.collections.id == collection_id) + & (db.collections.tenant_id == tenant_id) + ) + .select() + .first() + ) + if not collection: + raise NotFound("Collection not found") + + response_data = CollectionResponse( + id=collection.id, + tenant_id=collection.tenant_id, + team_id=collection.team_id, + name=collection.name, + description=collection.description, + parent_id=collection.parent_id, + created_by=collection.created_by, + created_at=collection.created_at, + ) + + detail = CollectionDetailResponse(data=response_data) + return jsonify(detail.model_dump(mode="json")), 200 + + +@collections_bp.route("/collections/", methods=["PUT"]) +@auth_required +@require_scope("collections:write", "collections:admin") +async def update_collection(collection_id: int): + """ + Update collection details. + + Only the creator or tenant admin can update. + Validates that parent_id changes don't create cycles. + + Requires: collections:write or collections:admin scope + """ + data = await request.get_json() + if not data: + raise BadRequest("Request body required") + + try: + req = CollectionUpdateRequest(**data) + except BadRequest as e: + # Validation error from field validators + error_msg = str(e).replace("400 Bad Request: ", "") + return jsonify({"error": error_msg}), 400 + except ValidationError as e: + return jsonify({"error": e.errors()[0].get("msg", "Validation error")}), 400 + + db = get_db() + user_id = g.current_user["id"] + tenant_id = g.current_user.get("tenant_id") + + if not tenant_id: + default_tenant = db(db.tenants.is_default == True).select().first() # noqa + if not default_tenant: + raise BadRequest("No tenant found for user") + tenant_id = default_tenant.id + + collection = ( + db( + (db.collections.id == collection_id) + & (db.collections.tenant_id == tenant_id) + ) + .select() + .first() + ) + if not collection: + raise NotFound("Collection not found") + + # Check authorization + if not _is_collection_member_or_admin(db, user_id, collection_id, tenant_id): + raise Forbidden("Cannot update this collection") + + # Validate parent_id if provided + if req.parent_id is not None: + try: + _validate_no_cycles(db, collection_id, req.parent_id) + except BadRequest as e: + error_msg = str(e).replace("400 Bad Request: ", "") + return jsonify({"error": error_msg}), 400 + + # Verify parent exists and belongs to same tenant + if req.parent_id: + parent = db( + (db.collections.id == req.parent_id) + & (db.collections.tenant_id == tenant_id) + ).select().first() + if not parent: + raise BadRequest("Invalid parent collection ID") + + # Update only provided fields + updates = {} + if req.name is not None: + updates["name"] = req.name + if req.description is not None: + updates["description"] = req.description + if req.parent_id is not None: + updates["parent_id"] = req.parent_id + + if updates: + db(db.collections.id == collection_id).update(**updates) + db.commit() + + # Fetch updated record + collection = db(db.collections.id == collection_id).select().first() + response_data = CollectionResponse( + id=collection.id, + tenant_id=collection.tenant_id, + team_id=collection.team_id, + name=collection.name, + description=collection.description, + parent_id=collection.parent_id, + created_by=collection.created_by, + created_at=collection.created_at, + ) + + return jsonify({"data": response_data.model_dump(mode="json")}), 200 + + +@collections_bp.route("/collections/", methods=["DELETE"]) +@auth_required +@require_scope("collections:delete", "collections:admin") +async def delete_collection(collection_id: int): + """ + Delete a collection (cascades to child collections and URLs). + + Only the creator or tenant admin can delete. + + Requires: collections:delete or collections:admin scope + """ + db = get_db() + user_id = g.current_user["id"] + tenant_id = g.current_user.get("tenant_id") + + if not tenant_id: + default_tenant = db(db.tenants.is_default == True).select().first() # noqa + if not default_tenant: + raise BadRequest("No tenant found for user") + tenant_id = default_tenant.id + + collection = ( + db( + (db.collections.id == collection_id) + & (db.collections.tenant_id == tenant_id) + ) + .select() + .first() + ) + if not collection: + raise NotFound("Collection not found") + + # Check authorization + if not _is_collection_member_or_admin(db, user_id, collection_id, tenant_id): + raise Forbidden("Cannot delete this collection") + + # Delete (cascade via ON DELETE CASCADE in schema) + db(db.collections.id == collection_id).delete() + db.commit() + + return jsonify({"message": "Collection deleted"}), 200 diff --git a/services/flask-backend/app/models.py b/services/flask-backend/app/models.py index 354368d8..e87e45e6 100644 --- a/services/flask-backend/app/models.py +++ b/services/flask-backend/app/models.py @@ -61,8 +61,10 @@ def init_db(app: Quart) -> DAL: # Enable WAL mode for SQLite to allow concurrent readers + one writer # without full table locking. Critical for multi-worker deployments. + # Also enable foreign key constraints for referential integrity. if "sqlite" in db_uri: try: + db.executesql("PRAGMA foreign_keys=ON") db.executesql("PRAGMA journal_mode=WAL") db.executesql("PRAGMA busy_timeout=30000") db.executesql("PRAGMA synchronous=NORMAL") @@ -155,6 +157,57 @@ def init_db(app: Quart) -> DAL: migrate=False, ) + # Define URL shortener tables for runtime use + if "collections" not in db.tables: + db.define_table( + "collections", + Field("tenant_id", "reference tenants"), + Field("team_id", "reference teams"), + Field("name", "string", length=255), + Field("description", "text"), + Field("parent_id", "reference collections"), + Field("created_by", "reference auth_user"), + Field("created_at", "datetime"), + migrate=False, + ) + + if "urls" not in db.tables: + db.define_table( + "urls", + Field("tenant_id", "reference tenants"), + Field("team_id", "reference teams"), + Field("collection_id", "reference collections"), + Field("short_code", "string", length=255), + Field("long_url", "text"), + Field("title", "string", length=255), + Field("description", "text"), + Field("is_active", "boolean", default=True), + Field("expires_at", "datetime"), + Field("click_count", "integer", default=0), + Field("created_by", "reference auth_user"), + Field("created_at", "datetime"), + Field("updated_at", "datetime"), + migrate=False, + ) + + if "link_clicks" not in db.tables: + db.define_table( + "link_clicks", + Field("url_id", "reference urls"), + Field("clicked_at", "datetime"), + Field("ip_hash", "string", length=64), + Field("user_agent", "text"), + Field("referer", "text"), + Field("country", "string", length=2), + Field("region", "string", length=255), + Field("city", "string", length=255), + Field("device_type", "string", length=50), + Field("browser", "string", length=100), + Field("os", "string", length=100), + Field("response_ms", "integer"), + migrate=False, + ) + # Ensure default roles exist _ensure_default_roles(db) @@ -344,6 +397,48 @@ def _create_tables_if_needed(db: DAL) -> None: external_email VARCHAR(255), linked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP )""", + """CREATE TABLE IF NOT EXISTS collections ( + id SERIAL PRIMARY KEY, + tenant_id INTEGER NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + team_id INTEGER REFERENCES teams(id) ON DELETE CASCADE, + name VARCHAR(255) NOT NULL, + description TEXT, + parent_id INTEGER REFERENCES collections(id) ON DELETE CASCADE, + created_by INTEGER REFERENCES auth_user(id), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + UNIQUE(tenant_id, name, parent_id) + )""", + """CREATE TABLE IF NOT EXISTS urls ( + id SERIAL PRIMARY KEY, + tenant_id INTEGER NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + team_id INTEGER REFERENCES teams(id) ON DELETE CASCADE, + collection_id INTEGER REFERENCES collections(id) ON DELETE SET NULL, + short_code VARCHAR(255) NOT NULL UNIQUE, + long_url TEXT NOT NULL, + title VARCHAR(255), + description TEXT, + is_active BOOLEAN DEFAULT TRUE, + expires_at TIMESTAMP, + click_count INTEGER DEFAULT 0, + created_by INTEGER REFERENCES auth_user(id), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + )""", + """CREATE TABLE IF NOT EXISTS link_clicks ( + id SERIAL PRIMARY KEY, + url_id INTEGER NOT NULL REFERENCES urls(id) ON DELETE CASCADE, + clicked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + ip_hash VARCHAR(64), + user_agent TEXT, + referer TEXT, + country VARCHAR(2), + region VARCHAR(255), + city VARCHAR(255), + device_type VARCHAR(50), + browser VARCHAR(100), + os VARCHAR(100), + response_ms INTEGER + )""", ]: try: db.executesql(sql) @@ -500,6 +595,48 @@ def _exec_ddl(sql: str) -> None: external_email VARCHAR(255), linked_at DATETIME DEFAULT CURRENT_TIMESTAMP )""", + """CREATE TABLE IF NOT EXISTS collections ( + id INTEGER PRIMARY KEY AUTO_INCREMENT, + tenant_id INTEGER NOT NULL REFERENCES tenants(id), + team_id INTEGER REFERENCES teams(id), + name VARCHAR(255) NOT NULL, + description TEXT, + parent_id INTEGER REFERENCES collections(id) ON DELETE CASCADE, + created_by INTEGER REFERENCES auth_user(id), + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + UNIQUE(tenant_id, name, parent_id) + )""", + """CREATE TABLE IF NOT EXISTS urls ( + id INTEGER PRIMARY KEY AUTO_INCREMENT, + tenant_id INTEGER NOT NULL REFERENCES tenants(id), + team_id INTEGER REFERENCES teams(id), + collection_id INTEGER REFERENCES collections(id) ON DELETE SET NULL, + short_code VARCHAR(255) NOT NULL UNIQUE, + long_url TEXT NOT NULL, + title VARCHAR(255), + description TEXT, + is_active BOOLEAN DEFAULT TRUE, + expires_at DATETIME, + click_count INTEGER DEFAULT 0, + created_by INTEGER REFERENCES auth_user(id), + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP + )""", + """CREATE TABLE IF NOT EXISTS link_clicks ( + id INTEGER PRIMARY KEY AUTO_INCREMENT, + url_id INTEGER NOT NULL REFERENCES urls(id), + clicked_at DATETIME DEFAULT CURRENT_TIMESTAMP, + ip_hash VARCHAR(64), + user_agent TEXT, + referer TEXT, + country VARCHAR(2), + region VARCHAR(255), + city VARCHAR(255), + device_type VARCHAR(50), + browser VARCHAR(100), + os VARCHAR(100), + response_ms INTEGER + )""", ]: _exec_ddl(sql) except Exception as e: diff --git a/services/flask-backend/app/rbac.py b/services/flask-backend/app/rbac.py index 58315f56..a0dc8398 100644 --- a/services/flask-backend/app/rbac.py +++ b/services/flask-backend/app/rbac.py @@ -40,6 +40,10 @@ # Analytics scopes "analytics:read": "Read analytics data", "analytics:admin": "Configure analytics settings", + # Collections scopes + "collections:read": "Read collections", + "collections:write": "Create and update collections", + "collections:delete": "Delete collections", # Settings scopes "settings:read": "Read application settings", "settings:write": "Modify application settings", @@ -64,6 +68,9 @@ "urls:write", "urls:delete", "urls:admin", + "collections:read", + "collections:write", + "collections:delete", "analytics:read", "analytics:admin", "settings:read", @@ -79,6 +86,8 @@ "urls:read", "urls:write", "urls:delete", + "collections:read", + "collections:write", "analytics:read", "settings:read", ], @@ -88,6 +97,7 @@ "teams:read", "tenants:read", "urls:read", + "collections:read", "analytics:read", "settings:read", ], @@ -103,6 +113,9 @@ "urls:read", "urls:write", "urls:delete", + "collections:read", + "collections:write", + "collections:delete", "analytics:read", ], "team_maintainer": [ @@ -110,12 +123,15 @@ "teams:read", "urls:read", "urls:write", + "collections:read", + "collections:write", "analytics:read", ], "team_viewer": [ "users:read", "teams:read", "urls:read", + "collections:read", "analytics:read", ], } @@ -133,6 +149,9 @@ "urls:read", "urls:write", "urls:delete", + "collections:read", + "collections:write", + "collections:delete", "analytics:read", ], "tenant_maintainer": [ @@ -141,6 +160,8 @@ "tenants:read", "urls:read", "urls:write", + "collections:read", + "collections:write", "analytics:read", ], "tenant_viewer": [ @@ -148,6 +169,7 @@ "teams:read", "tenants:read", "urls:read", + "collections:read", "analytics:read", ], } @@ -158,15 +180,21 @@ "urls:read", "urls:write", "urls:delete", + "collections:read", + "collections:write", + "collections:delete", "analytics:read", ], "editor": [ "urls:read", "urls:write", + "collections:read", + "collections:write", "analytics:read", ], "viewer": [ "urls:read", + "collections:read", "analytics:read", ], } diff --git a/services/flask-backend/app/schemas/collections.py b/services/flask-backend/app/schemas/collections.py new file mode 100644 index 00000000..c16374f3 --- /dev/null +++ b/services/flask-backend/app/schemas/collections.py @@ -0,0 +1,67 @@ +"""Collection management Pydantic models.""" + +from __future__ import annotations + +from datetime import datetime +from typing import List, Optional + +from penguin_libs.pydantic import ( + Description1000, + ImmutableModel, + Name255, + RequestModel, +) +from pydantic import Field, field_validator + + +class CollectionCreateRequest(RequestModel): + """Create collection request.""" + + name: Name255 = Field(..., description="Collection name") + description: Optional[str] = Field( + None, max_length=1000, description="Collection description" + ) + parent_id: Optional[int] = Field( + None, description="Parent collection ID for nested folders" + ) + team_id: Optional[int] = Field(None, description="Team ID (optional)") + + +class CollectionUpdateRequest(RequestModel): + """Update collection request.""" + + name: Optional[Name255] = Field(None, description="Updated collection name") + description: Optional[str] = Field( + None, max_length=1000, description="Updated description" + ) + parent_id: Optional[int] = Field( + None, description="Updated parent collection ID" + ) + + +class CollectionResponse(ImmutableModel): + """Single collection response.""" + + id: int = Field(..., description="Collection ID") + tenant_id: int = Field(..., description="Tenant ID") + team_id: Optional[int] = Field(None, description="Team ID") + name: str = Field(..., description="Collection name") + description: Optional[str] = Field(None, description="Description") + parent_id: Optional[int] = Field(None, description="Parent collection ID") + created_by: Optional[int] = Field(None, description="Creator user ID") + created_at: Optional[datetime] = Field(None, description="Creation time") + + +class CollectionListResponse(ImmutableModel): + """List of collections response.""" + + data: List[CollectionResponse] = Field(..., description="List of collections") + total: Optional[int] = Field(None, description="Total count (with pagination)") + limit: Optional[int] = Field(None, description="Limit used in query") + offset: Optional[int] = Field(None, description="Offset used in query") + + +class CollectionDetailResponse(ImmutableModel): + """Detailed collection response.""" + + data: CollectionResponse = Field(..., description="Collection details") diff --git a/services/flask-backend/app/schemas/urls.py b/services/flask-backend/app/schemas/urls.py new file mode 100644 index 00000000..df0d64bf --- /dev/null +++ b/services/flask-backend/app/schemas/urls.py @@ -0,0 +1,109 @@ +"""URL shortener Pydantic models.""" + +from __future__ import annotations + +from datetime import datetime +from typing import List, Optional + +from penguin_libs.pydantic import ( + Description1000, + ImmutableModel, + Name255, + RequestModel, +) +from pydantic import Field, field_validator + + +class URLCreateRequest(RequestModel): + """Create shortened URL request.""" + + long_url: str = Field(..., description="The long URL to shorten") + short_code: Optional[str] = Field( + None, + max_length=32, + description="Optional custom short code (alphanumeric + hyphens/underscores, max 32 chars)", + ) + title: Optional[str] = Field(None, max_length=255, description="URL title") + description: Optional[str] = Field( + None, max_length=1000, description="URL description" + ) + collection_id: Optional[int] = Field(None, description="Collection ID to organize link") + expires_at: Optional[datetime] = Field(None, description="Expiration timestamp") + team_id: Optional[int] = Field(None, description="Team ID (optional)") + + @field_validator("long_url", mode="before") + @classmethod + def validate_url(cls, v: str) -> str: + """Validate URL format and prevent SSRF.""" + if isinstance(v, str): + from app.urlvalidation import validate_destination_url + + validate_destination_url(v) + return v + + @field_validator("short_code", mode="before") + @classmethod + def validate_short_code_format(cls, v: Optional[str]) -> Optional[str]: + """Validate short code format if provided.""" + if v is not None: + from app.urlvalidation import validate_short_code + + validate_short_code(v) + return v + + +class URLUpdateRequest(RequestModel): + """Update shortened URL request.""" + + long_url: Optional[str] = Field(None, description="Updated long URL") + title: Optional[str] = Field(None, max_length=255, description="Updated title") + description: Optional[str] = Field( + None, max_length=1000, description="Updated description" + ) + collection_id: Optional[int] = Field(None, description="Updated collection ID") + is_active: Optional[bool] = Field(None, description="Active/inactive status") + expires_at: Optional[datetime] = Field(None, description="Updated expiration time") + + @field_validator("long_url", mode="before") + @classmethod + def validate_url(cls, v: Optional[str]) -> Optional[str]: + """Validate URL format if provided.""" + if v is not None: + from app.urlvalidation import validate_destination_url + + validate_destination_url(v) + return v + + +class URLResponse(ImmutableModel): + """Single URL response.""" + + id: int = Field(..., description="URL ID") + tenant_id: int = Field(..., description="Tenant ID") + team_id: Optional[int] = Field(None, description="Team ID") + collection_id: Optional[int] = Field(None, description="Collection ID") + short_code: str = Field(..., description="Short code") + long_url: str = Field(..., description="Long URL") + title: Optional[str] = Field(None, description="Title") + description: Optional[str] = Field(None, description="Description") + is_active: bool = Field(default=True, description="Active status") + expires_at: Optional[datetime] = Field(None, description="Expiration time") + click_count: int = Field(default=0, description="Number of clicks") + created_by: Optional[int] = Field(None, description="Creator user ID") + created_at: Optional[datetime] = Field(None, description="Creation time") + updated_at: Optional[datetime] = Field(None, description="Last update time") + + +class URLListResponse(ImmutableModel): + """List of URLs response.""" + + data: List[URLResponse] = Field(..., description="List of URLs") + total: Optional[int] = Field(None, description="Total count (with pagination)") + limit: Optional[int] = Field(None, description="Limit used in query") + offset: Optional[int] = Field(None, description="Offset used in query") + + +class URLDetailResponse(ImmutableModel): + """Detailed URL response with all fields.""" + + data: URLResponse = Field(..., description="URL details") diff --git a/services/flask-backend/app/urls.py b/services/flask-backend/app/urls.py new file mode 100644 index 00000000..ed94e0fe --- /dev/null +++ b/services/flask-backend/app/urls.py @@ -0,0 +1,468 @@ +""" +URL Shortener Management API Endpoints. + +Implements URL CRUD operations with scope-based permissions. +All operations are tenant-scoped. +""" + +from __future__ import annotations + +import secrets +import string +from datetime import datetime, timezone +from typing import Optional + +from pydantic import ValidationError +from quart import Blueprint, g, jsonify, request +from werkzeug.exceptions import BadRequest, Conflict, Forbidden, NotFound + +from .auth import auth_required +from .models import get_db +from .rbac import get_user_scopes, require_scope +from .schemas.urls import ( + URLCreateRequest, + URLDetailResponse, + URLListResponse, + URLResponse, + URLUpdateRequest, +) +from .urlvalidation import validate_destination_url, validate_short_code + +urls_bp = Blueprint("urls", __name__) + + +def _generate_short_code(length: int = 7) -> str: + """ + Generate a random short code using secrets module (cryptographically secure). + + Args: + length: Length of the code to generate + + Returns: + A random alphanumeric short code + """ + alphabet = string.ascii_letters + string.digits + return "".join(secrets.choice(alphabet) for _ in range(length)) + + +def _is_url_member_or_admin( + db, user_id: int, url_id: int, tenant_id: int +) -> bool: + """ + Check if user is the creator of the URL or is a tenant admin. + + Args: + db: Database instance + user_id: Current user ID + url_id: URL ID to check + tenant_id: Tenant ID for the URL + + Returns: + True if user created the URL or is tenant admin + """ + url = db(db.urls.id == url_id).select().first() + if not url: + return False + + # Creator can edit own links + if url.created_by == user_id: + return True + + # Tenant admin can edit any link in their tenant + from .rbac import _has_tenant_admin_scope + + return _has_tenant_admin_scope(db, user_id, tenant_id) + + +@urls_bp.route("/urls", methods=["POST"]) +@auth_required +@require_scope("urls:write", "urls:admin") +async def create_url(): + """ + Create a new shortened URL. + + Requires: urls:write or urls:admin scope + """ + data = await request.get_json() + if not data: + raise BadRequest("Request body required") + + try: + req = URLCreateRequest(**data) + except BadRequest as e: + # Validation error from field validators (e.g., SSRF checks) + error_msg = str(e).replace("400 Bad Request: ", "") + return jsonify({"error": error_msg}), 400 + except ValidationError as e: + return jsonify({"error": e.errors()[0].get("msg", "Validation error")}), 400 + + db = get_db() + user_id = g.current_user["id"] + tenant_id = g.current_user.get("tenant_id") + + if not tenant_id: + # Fallback: get default tenant + default_tenant = db(db.tenants.is_default == True).select().first() # noqa + if not default_tenant: + raise BadRequest("No tenant found for user") + tenant_id = default_tenant.id + + # Verify collection exists and belongs to this tenant if specified + if req.collection_id: + collection = db(db.collections.id == req.collection_id).select().first() + if not collection or collection.tenant_id != tenant_id: + raise BadRequest("Invalid collection ID") + + # Verify team if specified + if req.team_id: + team = db(db.teams.id == req.team_id).select().first() + if not team or team.tenant_id != tenant_id: + raise BadRequest("Invalid team ID") + + # Generate short code or use custom alias + if req.short_code: + # Custom alias provided - validate it + validate_short_code(req.short_code) + short_code = req.short_code + else: + # Generate random code, retry on collision (bounded) + max_retries = 10 + short_code = None + for attempt in range(max_retries): + candidate = _generate_short_code() + # Check uniqueness + existing = db(db.urls.short_code == candidate).select().first() + if not existing: + short_code = candidate + break + + if not short_code: + return ( + jsonify( + { + "error": "Could not generate unique short code after multiple attempts" + } + ), + 500, + ) + + # Try to insert - catch IntegrityError on duplicate short_code + try: + url_id = db.urls.insert( + tenant_id=tenant_id, + team_id=req.team_id, + collection_id=req.collection_id, + short_code=short_code, + long_url=req.long_url, + title=req.title, + description=req.description, + is_active=True, + expires_at=req.expires_at, + click_count=0, + created_by=user_id, + ) + db.commit() + except Exception as e: + db.rollback() + if "UNIQUE constraint failed" in str(e) or "duplicate key" in str(e): + return jsonify({"error": "Short code already exists"}), 409 + raise + + url = db(db.urls.id == url_id).select().first() + response_data = URLResponse( + id=url.id, + tenant_id=url.tenant_id, + team_id=url.team_id, + collection_id=url.collection_id, + short_code=url.short_code, + long_url=url.long_url, + title=url.title, + description=url.description, + is_active=url.is_active, + expires_at=url.expires_at, + click_count=url.click_count, + created_by=url.created_by, + created_at=url.created_at, + updated_at=url.updated_at, + ) + + return jsonify({"data": response_data.model_dump(mode="json")}), 201 + + +@urls_bp.route("/urls", methods=["GET"]) +@auth_required +@require_scope("urls:read") +async def list_urls(): + """ + List shortened URLs with pagination and filtering. + + Query params: + - limit: Items per page (default 20, max 100) + - offset: Pagination offset (default 0) + - collection_id: Filter by collection + - is_active: Filter by active status (true/false) + - search: Search short_code/long_url/title + + Requires: urls:read scope + """ + db = get_db() + user_id = g.current_user["id"] + tenant_id = g.current_user.get("tenant_id") + + if not tenant_id: + default_tenant = db(db.tenants.is_default == True).select().first() # noqa + if not default_tenant: + raise BadRequest("No tenant found for user") + tenant_id = default_tenant.id + + # Pagination + try: + limit = int(request.args.get("limit", 20)) + offset = int(request.args.get("offset", 0)) + limit = min(limit, 100) # Cap at 100 + except ValueError: + raise BadRequest("Invalid limit/offset") + + # Build query - base: tenant-scoped + query = db.urls.tenant_id == tenant_id + + # Optional filters + collection_id = request.args.get("collection_id") + if collection_id: + try: + collection_id = int(collection_id) + query &= db.urls.collection_id == collection_id + except ValueError: + raise BadRequest("Invalid collection_id") + + is_active = request.args.get("is_active") + if is_active is not None: + is_active_bool = is_active.lower() == "true" + query &= db.urls.is_active == is_active_bool + + search = request.args.get("search") + if search: + from pydal.objects import Query + + search_query = ( + (db.urls.short_code.contains(search)) + | (db.urls.long_url.contains(search)) + | (db.urls.title.contains(search)) + ) + query &= search_query + + # Count total + total = db(query).count() + + # Fetch with pagination + rows = ( + db(query) + .select(orderby=~db.urls.created_at, limitby=(offset, offset + limit)) + .as_list() + ) + + data = [ + URLResponse( + id=r["id"], + tenant_id=r["tenant_id"], + team_id=r["team_id"], + collection_id=r["collection_id"], + short_code=r["short_code"], + long_url=r["long_url"], + title=r["title"], + description=r["description"], + is_active=r["is_active"], + expires_at=r["expires_at"], + click_count=r["click_count"], + created_by=r["created_by"], + created_at=r["created_at"], + updated_at=r["updated_at"], + ) + for r in rows + ] + + response = URLListResponse( + data=data, + total=total, + limit=limit, + offset=offset, + ) + return jsonify(response.model_dump(mode="json")), 200 + + +@urls_bp.route("/urls/", methods=["GET"]) +@auth_required +@require_scope("urls:read") +async def get_url(url_id: int): + """ + Get URL details by ID. + + Requires: urls:read scope (tenant-scoped) + """ + db = get_db() + tenant_id = g.current_user.get("tenant_id") + + if not tenant_id: + default_tenant = db(db.tenants.is_default == True).select().first() # noqa + if not default_tenant: + raise BadRequest("No tenant found for user") + tenant_id = default_tenant.id + + url = ( + db((db.urls.id == url_id) & (db.urls.tenant_id == tenant_id)) + .select() + .first() + ) + if not url: + raise NotFound("URL not found") + + response_data = URLResponse( + id=url.id, + tenant_id=url.tenant_id, + team_id=url.team_id, + collection_id=url.collection_id, + short_code=url.short_code, + long_url=url.long_url, + title=url.title, + description=url.description, + is_active=url.is_active, + expires_at=url.expires_at, + click_count=url.click_count, + created_by=url.created_by, + created_at=url.created_at, + updated_at=url.updated_at, + ) + + detail = URLDetailResponse(data=response_data) + return jsonify(detail.model_dump(mode="json")), 200 + + +@urls_bp.route("/urls/", methods=["PUT"]) +@auth_required +@require_scope("urls:write", "urls:admin") +async def update_url(url_id: int): + """ + Update URL details. + + Only the creator or tenant admin can update. + Re-validates long_url if provided. + + Requires: urls:write or urls:admin scope + """ + data = await request.get_json() + if not data: + raise BadRequest("Request body required") + + try: + req = URLUpdateRequest(**data) + except ValidationError as e: + return jsonify({"error": e.errors()[0].get("msg", "Validation error")}), 400 + + db = get_db() + user_id = g.current_user["id"] + tenant_id = g.current_user.get("tenant_id") + + if not tenant_id: + default_tenant = db(db.tenants.is_default == True).select().first() # noqa + if not default_tenant: + raise BadRequest("No tenant found for user") + tenant_id = default_tenant.id + + url = ( + db((db.urls.id == url_id) & (db.urls.tenant_id == tenant_id)) + .select() + .first() + ) + if not url: + raise NotFound("URL not found") + + # Check authorization: creator or tenant admin + if not _is_url_member_or_admin(db, user_id, url_id, tenant_id): + raise Forbidden("Cannot update this URL") + + # Validate collection if provided + if req.collection_id: + collection = db(db.collections.id == req.collection_id).select().first() + if not collection or collection.tenant_id != tenant_id: + raise BadRequest("Invalid collection ID") + + # Update only provided fields + updates = {} + if req.long_url is not None: + updates["long_url"] = req.long_url + if req.title is not None: + updates["title"] = req.title + if req.description is not None: + updates["description"] = req.description + if req.collection_id is not None: + updates["collection_id"] = req.collection_id + if req.is_active is not None: + updates["is_active"] = req.is_active + if req.expires_at is not None: + updates["expires_at"] = req.expires_at + + if updates: + updates["updated_at"] = datetime.now(timezone.utc) + db(db.urls.id == url_id).update(**updates) + db.commit() + + # Fetch updated record + url = db(db.urls.id == url_id).select().first() + response_data = URLResponse( + id=url.id, + tenant_id=url.tenant_id, + team_id=url.team_id, + collection_id=url.collection_id, + short_code=url.short_code, + long_url=url.long_url, + title=url.title, + description=url.description, + is_active=url.is_active, + expires_at=url.expires_at, + click_count=url.click_count, + created_by=url.created_by, + created_at=url.created_at, + updated_at=url.updated_at, + ) + + return jsonify({"data": response_data.model_dump(mode="json")}), 200 + + +@urls_bp.route("/urls/", methods=["DELETE"]) +@auth_required +@require_scope("urls:delete", "urls:admin") +async def delete_url(url_id: int): + """ + Delete (soft delete) a URL. + + Only the creator or tenant admin can delete. + + Requires: urls:delete or urls:admin scope + """ + db = get_db() + user_id = g.current_user["id"] + tenant_id = g.current_user.get("tenant_id") + + if not tenant_id: + default_tenant = db(db.tenants.is_default == True).select().first() # noqa + if not default_tenant: + raise BadRequest("No tenant found for user") + tenant_id = default_tenant.id + + url = ( + db((db.urls.id == url_id) & (db.urls.tenant_id == tenant_id)) + .select() + .first() + ) + if not url: + raise NotFound("URL not found") + + # Check authorization + if not _is_url_member_or_admin(db, user_id, url_id, tenant_id): + raise Forbidden("Cannot delete this URL") + + # Soft delete: mark as inactive + db(db.urls.id == url_id).update(is_active=False) + db.commit() + + return jsonify({"message": "URL deleted"}), 200 diff --git a/services/flask-backend/app/urlvalidation.py b/services/flask-backend/app/urlvalidation.py new file mode 100644 index 00000000..43e8bc5e --- /dev/null +++ b/services/flask-backend/app/urlvalidation.py @@ -0,0 +1,219 @@ +""" +URL validation for preventing SSRF and open-redirect attacks. + +Validates destination URLs to prevent: +- Accessing private/loopback/reserved IP ranges +- Accessing cloud metadata endpoints (169.254.169.254) +- Using dangerous protocols (javascript:, data:, file:, etc.) +""" + +from __future__ import annotations + +import ipaddress +import socket +from typing import Optional +from urllib.parse import urlparse +from werkzeug.exceptions import BadRequest + +# Reserved paths that cannot be used as custom short codes +RESERVED_PATHS = { + "api", + "admin", + "health", + "healthz", + "readyz", + "login", + "logout", + "auth", + "static", + "assets", + "urls", + "collections", + "settings", + "users", + "teams", + "tenants", + "roles", + "docs", + "swagger", + "openapi", +} + +# Allowed URL schemes +ALLOWED_SCHEMES = {"http", "https"} + + +def validate_destination_url(url: str) -> None: + """ + Validate a destination URL to prevent SSRF and open-redirect attacks. + + Checks: + - URL scheme is http or https only + - Hostname does not resolve to private/loopback/reserved/multicast IP + - Cloud metadata IP (169.254.169.254) is rejected + - IPv6 addresses are properly validated + + Note: Resolution failures for hostnames are allowed (the shortener redirects + in the user's browser, not fetched server-side). Only literal IPs are checked + for private/reserved ranges without resolution. + + Args: + url: The URL to validate + + Raises: + BadRequest: If the URL fails validation + """ + import logging + logger = logging.getLogger(__name__) + + if not url: + raise BadRequest("URL cannot be empty") + + try: + parsed = urlparse(url) + except Exception as e: + raise BadRequest(f"Invalid URL: {e}") + + # Validate scheme + if not parsed.scheme: + raise BadRequest("URL must include a scheme (http or https)") + + if parsed.scheme.lower() not in ALLOWED_SCHEMES: + raise BadRequest( + f'Invalid URL scheme "{parsed.scheme}". Only http and https are allowed. scheme validation failed' + ) + + # Validate hostname + hostname = parsed.hostname + if not hostname: + raise BadRequest("URL must include a hostname") + + # Special handling for localhost hostname (not an IP address) + if hostname.lower() == "localhost": + raise BadRequest( + f'Hostname "{hostname}" is reserved loopback and cannot be used' + ) + + # Try to parse as IP address first (literal IP) + try: + ip = ipaddress.ip_address(hostname) + # Explicitly check for cloud metadata endpoint first (before other checks) + if ip == ipaddress.ip_address("169.254.169.254"): + raise BadRequest( + "Destination IP is cloud metadata endpoint and cannot be used" + ) + # Check for specific categories before generic "private" check + # (link-local, loopback, reserved, and multicast are subsets of private for some IP types) + if ip.is_loopback: + raise BadRequest( + f'Destination IP "{ip}" is loopback and cannot be used' + ) + if ip.is_link_local: + raise BadRequest( + f'Destination IP "{ip}" is link-local and cannot be used' + ) + if ip.is_multicast: + raise BadRequest( + f'Destination IP "{ip}" is multicast and cannot be used' + ) + if ip.is_reserved: + raise BadRequest( + f'Destination IP "{ip}" is reserved and cannot be used' + ) + if ip.is_private: + raise BadRequest( + f'Destination IP "{ip}" is private and cannot be used' + ) + # Accept valid public literal IP + return + except ValueError: + # Not a literal IP, continue to resolve as hostname + pass + + # Try to resolve hostname - if it fails, allow it (redirect doesn't fetch) + try: + results = socket.getaddrinfo(hostname, None) + if not results: + logger.warning(f"Could not resolve hostname: {hostname} (allowing)") + return + + for family, socktype, proto, canonname, sockaddr in results: + # sockaddr is (host, port) for IPv4 or (host, port, flowinfo, scopeid) for IPv6 + ip_str = sockaddr[0] + + try: + ip = ipaddress.ip_address(ip_str) + except ValueError as e: + raise BadRequest(f"Invalid IP address: {ip_str}") + + # Explicitly check for cloud metadata endpoint first (before other checks) + if ip == ipaddress.ip_address("169.254.169.254"): + raise BadRequest( + "Resolved IP is cloud metadata endpoint and cannot be used" + ) + + # Check for specific categories before generic "private" check + # (link-local, loopback, reserved, and multicast are subsets of private for some IP types) + if ip.is_loopback: + raise BadRequest( + f'Resolved IP "{ip_str}" is loopback and cannot be used' + ) + if ip.is_link_local: + raise BadRequest( + f'Resolved IP "{ip_str}" is link-local and cannot be used' + ) + if ip.is_multicast: + raise BadRequest( + f'Resolved IP "{ip_str}" is multicast and cannot be used' + ) + if ip.is_reserved: + raise BadRequest( + f'Resolved IP "{ip_str}" is reserved and cannot be used' + ) + if ip.is_private: + raise BadRequest( + f'Resolved IP "{ip_str}" is private and cannot be used' + ) + + except socket.gaierror: + # Resolution failed (NXDOMAIN, offline, etc.) + # This is OK for a shortener - we don't fetch the URL + logger.warning(f"Could not resolve hostname: {hostname} (allowing)") + return + except BadRequest: + raise + except socket.error as e: + raise BadRequest(f"Error validating URL hostname: {e}") + + +def validate_short_code(code: str) -> None: + """ + Validate a custom short code. + + Checks: + - Alphanumeric + hyphens and underscores only + - Length <= 32 + - Not in reserved paths + + Args: + code: The short code to validate + + Raises: + BadRequest: If the code fails validation + """ + if not code: + raise BadRequest("Short code cannot be empty") + + if len(code) > 32: + raise BadRequest("Short code must be 32 characters or less") + + # Allow alphanumeric, hyphens, underscores + import re + + if not re.match(r"^[a-zA-Z0-9_-]+$", code): + raise BadRequest( + "Short code must contain only alphanumeric characters, hyphens, and underscores" + ) + + if code.lower() in RESERVED_PATHS: + raise BadRequest(f"Short code '{code}' is reserved and cannot be used") diff --git a/services/flask-backend/tests/test_collections.py b/services/flask-backend/tests/test_collections.py new file mode 100644 index 00000000..26bfde96 --- /dev/null +++ b/services/flask-backend/tests/test_collections.py @@ -0,0 +1,530 @@ +"""Tests for /api/v1/collections endpoints.""" + +from __future__ import annotations + +import pytest + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +async def create_test_collection( + client, admin_headers, name: str, description: str = "", parent_id: int | None = None +) -> dict: + """Create a collection and return the parsed JSON.""" + payload = {"name": name, "description": description} + if parent_id: + payload["parent_id"] = parent_id + + resp = await client.post( + "/api/v1/collections", + json=payload, + headers=admin_headers, + ) + data = await resp.get_json() + assert resp.status_code == 201, f"create_test_collection failed ({resp.status_code}): {data}" + return data["data"] + + +# --------------------------------------------------------------------------- +# POST /api/v1/collections — create collection +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_create_collection_success(client, admin_headers): + """Admin can create a collection.""" + resp = await client.post( + "/api/v1/collections", + json={ + "name": "My Collection", + "description": "A test collection", + }, + headers=admin_headers, + ) + assert resp.status_code == 201 + data = await resp.get_json() + collection = data["data"] + assert collection["name"] == "My Collection" + assert collection["description"] == "A test collection" + assert collection["parent_id"] is None + + +@pytest.mark.asyncio +async def test_create_nested_collection(client, admin_headers): + """Can create nested collections (child of another collection).""" + parent = await create_test_collection(client, admin_headers, "Parent Collection") + + resp = await client.post( + "/api/v1/collections", + json={ + "name": "Child Collection", + "description": "Child of parent", + "parent_id": parent["id"], + }, + headers=admin_headers, + ) + assert resp.status_code == 201 + data = await resp.get_json() + collection = data["data"] + assert collection["name"] == "Child Collection" + assert collection["parent_id"] == parent["id"] + + +@pytest.mark.asyncio +async def test_create_collection_requires_auth(client): + """Unauthenticated request returns 401.""" + resp = await client.post( + "/api/v1/collections", + json={"name": "Unauthorized Collection"}, + ) + assert resp.status_code == 401 + + +@pytest.mark.asyncio +async def test_create_collection_duplicate_name_same_parent(client, admin_headers): + """Cannot create collection with same name in same parent scope.""" + await create_test_collection(client, admin_headers, "Unique Name") + + # Try to create another with same name + resp = await client.post( + "/api/v1/collections", + json={"name": "Unique Name"}, + headers=admin_headers, + ) + assert resp.status_code == 409 + data = await resp.get_json() + assert "already exists" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_create_collection_invalid_parent(client, admin_headers): + """Creating collection with non-existent parent fails.""" + resp = await client.post( + "/api/v1/collections", + json={ + "name": "Orphaned Collection", + "parent_id": 99999, + }, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "invalid" in data["error"].lower() or "not found" in data["error"].lower() + + +# --------------------------------------------------------------------------- +# GET /api/v1/collections — list collections +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_list_collections_empty(client, admin_headers): + """Can list collections.""" + resp = await client.get("/api/v1/collections", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + assert "data" in data + assert isinstance(data["data"], list) + + +@pytest.mark.asyncio +async def test_list_collections_with_pagination(client, admin_headers): + """List supports pagination.""" + for i in range(3): + await create_test_collection(client, admin_headers, f"Collection {i}") + + resp = await client.get( + "/api/v1/collections?limit=2&offset=0", headers=admin_headers + ) + assert resp.status_code == 200 + data = await resp.get_json() + assert len(data["data"]) <= 2 + assert data.get("limit") == 2 + + +@pytest.mark.asyncio +async def test_list_collections_filter_by_parent(client, admin_headers): + """Can filter collections by parent_id.""" + parent = await create_test_collection(client, admin_headers, "Parent") + child1 = await create_test_collection( + client, admin_headers, "Child 1", parent_id=parent["id"] + ) + child2 = await create_test_collection( + client, admin_headers, "Child 2", parent_id=parent["id"] + ) + + # List children of parent + resp = await client.get( + f"/api/v1/collections?parent_id={parent['id']}", headers=admin_headers + ) + assert resp.status_code == 200 + data = await resp.get_json() + # Should include both children + names = [c["name"] for c in data["data"]] + assert "Child 1" in names + assert "Child 2" in names + + +# --------------------------------------------------------------------------- +# GET /api/v1/collections/ — get single collection +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_get_collection_success(client, admin_headers): + """Can retrieve collection by ID.""" + created = await create_test_collection(client, admin_headers, "Test Collection") + collection_id = created["id"] + + resp = await client.get(f"/api/v1/collections/{collection_id}", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + collection = data["data"] + assert collection["id"] == collection_id + assert collection["name"] == "Test Collection" + + +@pytest.mark.asyncio +async def test_get_collection_not_found(client, admin_headers): + """Getting non-existent collection returns 404.""" + resp = await client.get("/api/v1/collections/99999", headers=admin_headers) + assert resp.status_code == 404 + + +# --------------------------------------------------------------------------- +# PUT /api/v1/collections/ — update collection +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_update_collection_success(client, admin_headers): + """Can update collection details.""" + created = await create_test_collection(client, admin_headers, "Original Name") + + resp = await client.put( + f"/api/v1/collections/{created['id']}", + json={"name": "Updated Name", "description": "Updated description"}, + headers=admin_headers, + ) + assert resp.status_code == 200 + data = await resp.get_json() + collection = data["data"] + assert collection["name"] == "Updated Name" + assert collection["description"] == "Updated description" + + +@pytest.mark.asyncio +async def test_update_collection_change_parent(client, admin_headers): + """Can move collection to different parent.""" + old_parent = await create_test_collection(client, admin_headers, "Old Parent") + new_parent = await create_test_collection(client, admin_headers, "New Parent") + child = await create_test_collection( + client, admin_headers, "Child", parent_id=old_parent["id"] + ) + + resp = await client.put( + f"/api/v1/collections/{child['id']}", + json={"parent_id": new_parent["id"]}, + headers=admin_headers, + ) + assert resp.status_code == 200 + data = await resp.get_json() + collection = data["data"] + assert collection["parent_id"] == new_parent["id"] + + +@pytest.mark.asyncio +async def test_update_collection_prevent_self_parent(client, admin_headers): + """Cannot set a collection as its own parent.""" + collection = await create_test_collection(client, admin_headers, "Self Reference") + + resp = await client.put( + f"/api/v1/collections/{collection['id']}", + json={"parent_id": collection["id"]}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "parent" in data["error"].lower() and ("cannot" in data["error"].lower() or "cycle" in data["error"].lower()) + + +@pytest.mark.asyncio +async def test_update_collection_prevent_cycle(client, admin_headers): + """Cannot create a cycle in collection hierarchy.""" + # Create: A -> B -> C + # Then try: C.parent = A (which would create cycle) + parent_a = await create_test_collection(client, admin_headers, "Parent A") + parent_b = await create_test_collection( + client, admin_headers, "Parent B", parent_id=parent_a["id"] + ) + child_c = await create_test_collection( + client, admin_headers, "Child C", parent_id=parent_b["id"] + ) + + # Try to set C's parent to A's grandchild (creating cycle) + resp = await client.put( + f"/api/v1/collections/{parent_a['id']}", + json={"parent_id": child_c["id"]}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "cycle" in data["error"].lower() + + +# --------------------------------------------------------------------------- +# DELETE /api/v1/collections/ — delete collection +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_delete_collection_success(client, admin_headers): + """Can delete a collection.""" + created = await create_test_collection(client, admin_headers, "Deletable") + + resp = await client.delete( + f"/api/v1/collections/{created['id']}", headers=admin_headers + ) + assert resp.status_code == 200 + + # Verify collection is gone + resp = await client.get(f"/api/v1/collections/{created['id']}", headers=admin_headers) + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_delete_collection_cascades_to_children(client, admin_headers): + """Deleting a collection cascades to child collections.""" + parent = await create_test_collection(client, admin_headers, "Parent") + child = await create_test_collection( + client, admin_headers, "Child", parent_id=parent["id"] + ) + + # Delete parent + resp = await client.delete(f"/api/v1/collections/{parent['id']}", headers=admin_headers) + assert resp.status_code == 200 + + # Verify child is also gone (cascade delete) + resp = await client.get(f"/api/v1/collections/{child['id']}", headers=admin_headers) + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_delete_collection_not_found(client, admin_headers): + """Deleting non-existent collection returns 404.""" + resp = await client.delete("/api/v1/collections/99999", headers=admin_headers) + assert resp.status_code == 404 + + +# --------------------------------------------------------------------------- +# Tenant Isolation +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_collection_tenant_isolation(client, admin_headers, viewer_headers): + """Collections are tenant-scoped.""" + # Admin creates collection + admin_collection = await create_test_collection(client, admin_headers, "Admin Collection") + + # Viewer tries to access (behavior depends on tenant assignment) + resp = await client.get( + f"/api/v1/collections/{admin_collection['id']}", headers=viewer_headers + ) + # May return 404 if in different tenant, or 200 if same tenant + # At minimum, viewer shouldn't be able to update or delete if in different tenant + + +# --------------------------------------------------------------------------- +# Additional Coverage Tests +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_update_collection_name(client, admin_headers): + """Update collection name only.""" + collection = await create_test_collection(client, admin_headers, "Original Name") + + resp = await client.put( + f"/api/v1/collections/{collection['id']}", + json={"name": "Updated Name"}, + headers=admin_headers, + ) + assert resp.status_code == 200 + data = await resp.get_json() + assert data["data"]["name"] == "Updated Name" + + +@pytest.mark.asyncio +async def test_update_collection_description(client, admin_headers): + """Update collection description only.""" + collection = await create_test_collection(client, admin_headers, "Test Collection", "Old desc") + + resp = await client.put( + f"/api/v1/collections/{collection['id']}", + json={"description": "New description"}, + headers=admin_headers, + ) + assert resp.status_code == 200 + data = await resp.get_json() + assert data["data"]["description"] == "New description" + + +@pytest.mark.asyncio +async def test_get_collection_not_found(client, admin_headers): + """Getting non-existent collection returns 404.""" + resp = await client.get(f"/api/v1/collections/99999", headers=admin_headers) + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_update_collection_not_found(client, admin_headers): + """Updating non-existent collection returns 404.""" + resp = await client.put( + f"/api/v1/collections/99999", + json={"name": "New Name"}, + headers=admin_headers, + ) + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_list_collections_with_parent_filter_none(client, admin_headers): + """List collections filtered by parent_id=None (root collections only).""" + # Create root collections + root1 = await create_test_collection(client, admin_headers, "Root 1") + root2 = await create_test_collection(client, admin_headers, "Root 2") + + # Create nested collection + child = await create_test_collection(client, admin_headers, "Child", parent_id=root1["id"]) + + # Filter for root only + resp = await client.get("/api/v1/collections?parent_id=none", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + + # Should have at least our two root collections + root_names = {c["name"] for c in data["data"]} + assert "Root 1" in root_names or "Root 2" in root_names + + +@pytest.mark.asyncio +async def test_list_collections_filter_by_parent(client, admin_headers): + """List collections filtered by specific parent_id.""" + parent = await create_test_collection(client, admin_headers, "Parent") + child1 = await create_test_collection(client, admin_headers, "Child 1", parent_id=parent["id"]) + child2 = await create_test_collection(client, admin_headers, "Child 2", parent_id=parent["id"]) + + # Create another parent to make sure it's not included + other_parent = await create_test_collection(client, admin_headers, "Other Parent") + other_child = await create_test_collection(client, admin_headers, "Other Child", parent_id=other_parent["id"]) + + # Filter for children of parent + resp = await client.get(f"/api/v1/collections?parent_id={parent['id']}", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + + # Should only have children of parent + child_names = {c["name"] for c in data["data"]} + assert "Child 1" in child_names or "Child 2" in child_names + assert "Other Child" not in child_names + + +@pytest.mark.asyncio +async def test_list_collections_pagination(client, admin_headers): + """List collections supports pagination.""" + # Create multiple collections + for i in range(5): + await create_test_collection(client, admin_headers, f"Collection {i}") + + # Get first page with limit 2 + resp = await client.get("/api/v1/collections?limit=2&offset=0", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + assert len(data["data"]) <= 2 + assert data["limit"] == 2 + assert data["offset"] == 0 + + # Get second page + resp = await client.get("/api/v1/collections?limit=2&offset=2", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + assert data["limit"] == 2 + assert data["offset"] == 2 + + +@pytest.mark.asyncio +async def test_create_collection_no_body(client, admin_headers): + """Creating collection without request body fails.""" + resp = await client.post( + "/api/v1/collections", + json=None, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "error" in data + + +@pytest.mark.asyncio +async def test_create_collection_with_invalid_team(client, admin_headers): + """Creating collection with non-existent team fails.""" + resp = await client.post( + "/api/v1/collections", + json={ + "name": "Test", + "team_id": 99999, + }, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "team" in data["error"].lower() or "invalid" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_update_collection_with_invalid_parent(client, admin_headers): + """Updating collection with non-existent parent fails.""" + collection = await create_test_collection(client, admin_headers, "Test") + + resp = await client.put( + f"/api/v1/collections/{collection['id']}", + json={"parent_id": 99999}, + headers=admin_headers, + ) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_list_collections_invalid_limit(client, admin_headers): + """List with invalid limit parameter fails.""" + resp = await client.get("/api/v1/collections?limit=invalid", headers=admin_headers) + assert resp.status_code == 400 + data = await resp.get_json() + assert "error" in data + + +@pytest.mark.asyncio +async def test_list_collections_invalid_parent_id_param(client, admin_headers): + """List with invalid parent_id parameter fails.""" + resp = await client.get("/api/v1/collections?parent_id=invalid", headers=admin_headers) + assert resp.status_code == 400 + data = await resp.get_json() + assert "error" in data + + +@pytest.mark.asyncio +async def test_update_collection_no_body(client, admin_headers): + """Updating collection without request body fails.""" + collection = await create_test_collection(client, admin_headers, "Test") + resp = await client.put( + f"/api/v1/collections/{collection['id']}", + json=None, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "error" in data diff --git a/services/flask-backend/tests/test_urls.py b/services/flask-backend/tests/test_urls.py new file mode 100644 index 00000000..eee1076b --- /dev/null +++ b/services/flask-backend/tests/test_urls.py @@ -0,0 +1,803 @@ +"""Tests for /api/v1/urls endpoints.""" + +from __future__ import annotations + +import pytest + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +async def create_test_url( + client, admin_headers, long_url: str, short_code: str | None = None +) -> dict: + """Create a URL and return the parsed JSON.""" + payload = {"long_url": long_url} + if short_code: + payload["short_code"] = short_code + + resp = await client.post("/api/v1/urls", json=payload, headers=admin_headers) + data = await resp.get_json() + assert resp.status_code == 201, f"create_test_url failed ({resp.status_code}): {data}" + return data["data"] + + +async def create_test_collection( + client, admin_headers, name: str, description: str = "" +) -> dict: + """Create a collection and return the parsed JSON.""" + resp = await client.post( + "/api/v1/collections", + json={"name": name, "description": description}, + headers=admin_headers, + ) + data = await resp.get_json() + assert resp.status_code == 201, f"create_test_collection failed: {data}" + return data["data"] + + +# --------------------------------------------------------------------------- +# POST /api/v1/urls — create URL +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_create_url_success(client, admin_headers): + """Admin can create a shortened URL with generated code.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "https://www.example.com/very/long/url"}, + headers=admin_headers, + ) + assert resp.status_code == 201 + data = await resp.get_json() + url = data["data"] + assert url["long_url"] == "https://www.example.com/very/long/url" + assert url["short_code"] + assert len(url["short_code"]) == 7 # default length + assert url["is_active"] is True + + +@pytest.mark.asyncio +async def test_create_url_with_custom_alias(client, admin_headers): + """Admin can create URL with custom short code.""" + resp = await client.post( + "/api/v1/urls", + json={ + "long_url": "https://www.example.com/foo", + "short_code": "my-link", + }, + headers=admin_headers, + ) + assert resp.status_code == 201 + data = await resp.get_json() + url = data["data"] + assert url["short_code"] == "my-link" + + +@pytest.mark.asyncio +async def test_create_url_requires_auth(client): + """Unauthenticated request returns 401.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "https://www.example.com"}, + ) + assert resp.status_code == 401 + + +@pytest.mark.asyncio +async def test_create_url_with_title_and_description(client, admin_headers): + """Can create URL with title and description.""" + resp = await client.post( + "/api/v1/urls", + json={ + "long_url": "https://www.example.com", + "title": "My Link", + "description": "A test link", + }, + headers=admin_headers, + ) + assert resp.status_code == 201 + data = await resp.get_json() + url = data["data"] + assert url["title"] == "My Link" + assert url["description"] == "A test link" + + +@pytest.mark.asyncio +async def test_create_url_unique_short_code_enforcement(client, admin_headers): + """Creating URL with duplicate short code fails.""" + # Create first URL with custom code + resp1 = await client.post( + "/api/v1/urls", + json={ + "long_url": "https://www.example.com/first", + "short_code": "unique-code", + }, + headers=admin_headers, + ) + assert resp1.status_code == 201 + + # Try to create second with same code + resp2 = await client.post( + "/api/v1/urls", + json={ + "long_url": "https://www.example.com/second", + "short_code": "unique-code", + }, + headers=admin_headers, + ) + assert resp2.status_code == 409 + data = await resp2.get_json() + assert "already exists" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_create_url_reserved_alias_rejected(client, admin_headers): + """Using reserved path as alias fails validation.""" + reserved_words = ["api", "admin", "login", "health", "static"] + + for reserved in reserved_words: + resp = await client.post( + "/api/v1/urls", + json={ + "long_url": "https://www.example.com", + "short_code": reserved, + }, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "reserved" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_create_url_invalid_alias_charset(client, admin_headers): + """Alias with invalid characters fails validation.""" + resp = await client.post( + "/api/v1/urls", + json={ + "long_url": "https://www.example.com", + "short_code": "invalid@alias!", + }, + headers=admin_headers, + ) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_create_url_with_collection(client, admin_headers): + """Can create URL in a collection.""" + collection = await create_test_collection(client, admin_headers, "My Links") + + resp = await client.post( + "/api/v1/urls", + json={ + "long_url": "https://www.example.com", + "collection_id": collection["id"], + }, + headers=admin_headers, + ) + assert resp.status_code == 201 + data = await resp.get_json() + url = data["data"] + assert url["collection_id"] == collection["id"] + + +# --------------------------------------------------------------------------- +# SSRF / Open-Redirect Prevention Tests +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_ssrf_private_ipv4_rejected(client, admin_headers): + """URLs pointing to private IPv4 ranges are rejected.""" + private_urls = [ + "http://192.168.1.1/", + "http://10.0.0.1/", + "http://172.16.0.1/", + "http://127.0.0.1/", + ] + + for url in private_urls: + resp = await client.post( + "/api/v1/urls", + json={"long_url": url}, + headers=admin_headers, + ) + assert resp.status_code == 400, f"Should reject {url}" + data = await resp.get_json() + assert "private" in data["error"].lower() or "reserved" in data["error"].lower() or "loopback" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_ssrf_loopback_rejected(client, admin_headers): + """Loopback addresses are rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "http://localhost/"}, + headers=admin_headers, + ) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_ssrf_cloud_metadata_rejected(client, admin_headers): + """Cloud metadata endpoint is rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "http://169.254.169.254/latest/meta-data/"}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "metadata" in data["error"].lower() or "not allowed" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_ssrf_invalid_scheme_rejected(client, admin_headers): + """Non-http/https schemes are rejected.""" + dangerous_schemes = [ + "javascript:alert(1)", + "data:text/html,", + "file:///etc/passwd", + ] + + for url in dangerous_schemes: + resp = await client.post( + "/api/v1/urls", + json={"long_url": url}, + headers=admin_headers, + ) + assert resp.status_code == 400, f"Should reject {url}" + + +# --------------------------------------------------------------------------- +# GET /api/v1/urls — list URLs +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_list_urls_empty(client, admin_headers): + """Can list URLs (may be empty initially).""" + resp = await client.get("/api/v1/urls", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + assert "data" in data + assert isinstance(data["data"], list) + + +@pytest.mark.asyncio +async def test_list_urls_with_pagination(client, admin_headers): + """List supports pagination via limit/offset.""" + # Create a few URLs + for i in range(3): + await create_test_url(client, admin_headers, f"https://example.com/{i}") + + resp = await client.get("/api/v1/urls?limit=2&offset=0", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + assert len(data["data"]) <= 2 + assert data.get("limit") == 2 + assert data.get("offset") == 0 + + +@pytest.mark.asyncio +async def test_list_urls_search_by_short_code(client, admin_headers): + """Can search URLs by short code.""" + url = await create_test_url(client, admin_headers, "https://example.com/search-test") + short_code = url["short_code"] + + resp = await client.get( + f"/api/v1/urls?search={short_code}", headers=admin_headers + ) + assert resp.status_code == 200 + data = await resp.get_json() + assert len(data["data"]) >= 1 + assert any(u["short_code"] == short_code for u in data["data"]) + + +@pytest.mark.asyncio +async def test_list_urls_filter_by_active(client, admin_headers): + """Can filter URLs by is_active status.""" + resp = await client.get("/api/v1/urls?is_active=true", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + # All URLs in response should be active + for url in data["data"]: + assert url["is_active"] is True + + +# --------------------------------------------------------------------------- +# GET /api/v1/urls/ — get single URL +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_get_url_success(client, admin_headers): + """Can retrieve URL by ID.""" + created = await create_test_url(client, admin_headers, "https://example.com") + url_id = created["id"] + + resp = await client.get(f"/api/v1/urls/{url_id}", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + url = data["data"] + assert url["id"] == url_id + + +@pytest.mark.asyncio +async def test_get_url_not_found(client, admin_headers): + """Getting non-existent URL returns 404.""" + resp = await client.get("/api/v1/urls/99999", headers=admin_headers) + assert resp.status_code == 404 + + +# --------------------------------------------------------------------------- +# PUT /api/v1/urls/ — update URL +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_update_url_success(client, admin_headers): + """Can update URL details.""" + created = await create_test_url(client, admin_headers, "https://example.com/old") + + resp = await client.put( + f"/api/v1/urls/{created['id']}", + json={"title": "Updated Title"}, + headers=admin_headers, + ) + assert resp.status_code == 200 + data = await resp.get_json() + url = data["data"] + assert url["title"] == "Updated Title" + + +@pytest.mark.asyncio +async def test_update_url_long_url_revalidated(client, admin_headers): + """Updating long_url re-validates for SSRF.""" + created = await create_test_url(client, admin_headers, "https://example.com") + + resp = await client.put( + f"/api/v1/urls/{created['id']}", + json={"long_url": "http://192.168.1.1/"}, + headers=admin_headers, + ) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_update_url_change_active_status(client, admin_headers): + """Can toggle is_active status.""" + created = await create_test_url(client, admin_headers, "https://example.com") + + resp = await client.put( + f"/api/v1/urls/{created['id']}", + json={"is_active": False}, + headers=admin_headers, + ) + assert resp.status_code == 200 + data = await resp.get_json() + url = data["data"] + assert url["is_active"] is False + + +# --------------------------------------------------------------------------- +# DELETE /api/v1/urls/ — delete URL +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_delete_url_success(client, admin_headers): + """Can soft-delete a URL.""" + created = await create_test_url(client, admin_headers, "https://example.com") + + resp = await client.delete(f"/api/v1/urls/{created['id']}", headers=admin_headers) + assert resp.status_code == 200 + + # Verify URL is marked inactive + resp = await client.get(f"/api/v1/urls/{created['id']}", headers=admin_headers) + data = await resp.get_json() + url = data["data"] + assert url["is_active"] is False + + +@pytest.mark.asyncio +async def test_delete_url_not_found(client, admin_headers): + """Deleting non-existent URL returns 404.""" + resp = await client.delete("/api/v1/urls/99999", headers=admin_headers) + assert resp.status_code == 404 + + +# --------------------------------------------------------------------------- +# Tenant Isolation +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_url_tenant_isolation(client, admin_headers, viewer_headers): + """Users in different tenants cannot access each other's URLs.""" + # Admin creates URL + admin_url = await create_test_url(client, admin_headers, "https://admin.example.com") + + # Viewer tries to access (may not have permission or different tenant) + # If viewer is in different tenant, should not see the URL + resp = await client.get(f"/api/v1/urls/{admin_url['id']}", headers=viewer_headers) + # Expect 404 if viewer is in different tenant + # (This depends on how tenants are assigned in test fixtures) + # At minimum, viewer shouldn't be able to update or delete admin's URL + + +# --------------------------------------------------------------------------- +# URL Validation Edge Cases (coverage for urlvalidation.py) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_url_empty_url_rejected(client, admin_headers): + """Empty URL is rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": ""}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "empty" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_url_no_scheme_rejected(client, admin_headers): + """URL without scheme is rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "www.example.com"}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "scheme" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_url_no_hostname_rejected(client, admin_headers): + """URL without hostname is rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "http://"}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "hostname" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_url_invalid_scheme_data(client, admin_headers): + """data: scheme is rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "data:text/html;base64,PHNjcmlwdD5hbGVydCgxKTwvc2NyaXB0Pg=="}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "scheme" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_url_invalid_scheme_file(client, admin_headers): + """file: scheme is rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "file:///etc/passwd"}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "scheme" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_url_ssrf_loopback_ipv6(client, admin_headers): + """IPv6 loopback address [::1] is rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "http://[::1]/"}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "loopback" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_url_ssrf_link_local_ipv6(client, admin_headers): + """IPv6 link-local address is rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "http://[fe80::1]/"}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "link-local" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_url_ssrf_multicast(client, admin_headers): + """Multicast address is rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "http://224.0.0.1/"}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "multicast" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_url_ssrf_reserved_ip(client, admin_headers): + """Reserved IP address is rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "http://255.255.255.255/"}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "reserved" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_url_valid_public_ip(client, admin_headers): + """Valid public IP addresses are allowed.""" + # 8.8.8.8 is Google's public DNS, should be allowed + resp = await client.post( + "/api/v1/urls", + json={"long_url": "http://8.8.8.8/"}, + headers=admin_headers, + ) + assert resp.status_code == 201 + + +@pytest.mark.asyncio +async def test_short_code_empty_rejected(client, admin_headers): + """Empty short code is rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "https://example.com", "short_code": ""}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "short code" in data["error"].lower() or "empty" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_short_code_too_long_rejected(client, admin_headers): + """Short code longer than 32 characters is rejected.""" + long_code = "a" * 33 + resp = await client.post( + "/api/v1/urls", + json={"long_url": "https://example.com", "short_code": long_code}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "32 characters" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_short_code_invalid_chars_rejected(client, admin_headers): + """Short code with invalid characters is rejected.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "https://example.com", "short_code": "test@code"}, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "alphanumeric" in data["error"].lower() + + +@pytest.mark.asyncio +async def test_url_list_filter_by_is_active(client, admin_headers): + """List URLs can be filtered by is_active status.""" + # Create an active URL + url1 = await create_test_url(client, admin_headers, "https://active.example.com") + + # Create another URL and deactivate it + url2 = await create_test_url(client, admin_headers, "https://inactive.example.com") + await client.put( + f"/api/v1/urls/{url2['id']}", + json={"is_active": False}, + headers=admin_headers, + ) + + # Filter for active URLs + resp = await client.get("/api/v1/urls?is_active=true", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + assert len(data["data"]) >= 1 + assert all(u["is_active"] for u in data["data"]) + + # Filter for inactive URLs + resp = await client.get("/api/v1/urls?is_active=false", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + assert any(not u["is_active"] for u in data["data"]) + + +@pytest.mark.asyncio +async def test_url_list_invalid_limit(client, admin_headers): + """Invalid limit parameter returns error.""" + resp = await client.get("/api/v1/urls?limit=invalid", headers=admin_headers) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_url_list_search_long_url(client, admin_headers): + """Search by long_url works.""" + await create_test_url(client, admin_headers, "https://search-test.example.com/path") + + resp = await client.get("/api/v1/urls?search=search-test", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + # Should find the URL + assert any("search-test" in u["long_url"] for u in data["data"]) + + +@pytest.mark.asyncio +async def test_url_without_title_and_description(client, admin_headers): + """URLs can be created without title and description.""" + resp = await client.post( + "/api/v1/urls", + json={"long_url": "https://minimal.example.com"}, + headers=admin_headers, + ) + assert resp.status_code == 201 + data = await resp.get_json() + assert data["data"]["title"] is None + assert data["data"]["description"] is None + + +@pytest.mark.asyncio +async def test_update_url_with_expiry(client, admin_headers): + """URL can be updated with expiry time.""" + url = await create_test_url(client, admin_headers, "https://expiry.example.com") + + # Update with future expiry + from datetime import datetime, timedelta, timezone + future = datetime.now(timezone.utc) + timedelta(days=1) + + resp = await client.put( + f"/api/v1/urls/{url['id']}", + json={"expires_at": future.isoformat()}, + headers=admin_headers, + ) + assert resp.status_code == 200 + data = await resp.get_json() + assert data["data"]["expires_at"] is not None + + +@pytest.mark.asyncio +async def test_delete_url_by_creator(client, admin_headers): + """Creator can delete their own URL.""" + url = await create_test_url(client, admin_headers, "https://delete-test.example.com") + + resp = await client.delete(f"/api/v1/urls/{url['id']}", headers=admin_headers) + assert resp.status_code == 200 + data = await resp.get_json() + assert "message" in data or "data" in data + + +@pytest.mark.asyncio +async def test_create_url_no_body(client, admin_headers): + """Creating URL without request body fails.""" + resp = await client.post( + "/api/v1/urls", + json=None, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "error" in data + + +@pytest.mark.asyncio +async def test_create_url_invalid_collection(client, admin_headers): + """Creating URL with non-existent collection fails.""" + resp = await client.post( + "/api/v1/urls", + json={ + "long_url": "https://example.com", + "collection_id": 99999, + }, + headers=admin_headers, + ) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_create_url_invalid_team(client, admin_headers): + """Creating URL with non-existent team fails.""" + resp = await client.post( + "/api/v1/urls", + json={ + "long_url": "https://example.com", + "team_id": 99999, + }, + headers=admin_headers, + ) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_get_url_not_found(client, admin_headers): + """Getting non-existent URL returns 404.""" + resp = await client.get("/api/v1/urls/99999", headers=admin_headers) + assert resp.status_code == 404 + data = await resp.get_json() + assert "error" in data or "not found" in str(data).lower() + + +@pytest.mark.asyncio +async def test_update_url_not_found(client, admin_headers): + """Updating non-existent URL returns 404.""" + resp = await client.put( + "/api/v1/urls/99999", + json={"title": "New Title"}, + headers=admin_headers, + ) + assert resp.status_code == 404 + data = await resp.get_json() + assert "error" in data or "not found" in str(data).lower() + + +@pytest.mark.asyncio +async def test_delete_url_not_found(client, admin_headers): + """Deleting non-existent URL returns 404.""" + resp = await client.delete("/api/v1/urls/99999", headers=admin_headers) + assert resp.status_code == 404 + data = await resp.get_json() + assert "error" in data or "not found" in str(data).lower() + + +@pytest.mark.asyncio +async def test_update_url_no_body(client, admin_headers): + """Updating URL without request body fails.""" + url = await create_test_url(client, admin_headers, "https://example.com") + resp = await client.put( + f"/api/v1/urls/{url['id']}", + json=None, + headers=admin_headers, + ) + assert resp.status_code == 400 + data = await resp.get_json() + assert "error" in data + + +@pytest.mark.asyncio +async def test_update_url_invalid_collection(client, admin_headers): + """Updating URL with non-existent collection fails.""" + url = await create_test_url(client, admin_headers, "https://example.com") + resp = await client.put( + f"/api/v1/urls/{url['id']}", + json={"collection_id": 99999}, + headers=admin_headers, + ) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_list_urls_invalid_collection_id(client, admin_headers): + """List with invalid collection_id parameter fails.""" + resp = await client.get("/api/v1/urls?collection_id=invalid", headers=admin_headers) + assert resp.status_code == 400 + data = await resp.get_json() + assert "error" in data diff --git a/services/flask-backend/tests/test_urlvalidation.py b/services/flask-backend/tests/test_urlvalidation.py new file mode 100644 index 00000000..2bc993a1 --- /dev/null +++ b/services/flask-backend/tests/test_urlvalidation.py @@ -0,0 +1,432 @@ +"""Comprehensive tests for URL validation module (SSRF/open-redirect prevention).""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from werkzeug.exceptions import BadRequest + +from app.urlvalidation import validate_destination_url, validate_short_code + + +# --------------------------------------------------------------------------- +# validate_destination_url Tests +# --------------------------------------------------------------------------- + + +class TestValidateDestinationUrlBasic: + """Basic URL validation tests.""" + + def test_empty_url_rejected(self) -> None: + """Empty URL is rejected.""" + with pytest.raises(BadRequest, match="cannot be empty"): + validate_destination_url("") + + def test_none_url_rejected(self) -> None: + """None/falsy URL is rejected.""" + with pytest.raises(BadRequest, match="cannot be empty"): + validate_destination_url(None) # type: ignore + + def test_url_without_scheme_rejected(self) -> None: + """URL without scheme is rejected.""" + with pytest.raises(BadRequest, match="must include a scheme"): + validate_destination_url("www.example.com") + + def test_url_with_invalid_scheme_rejected(self) -> None: + """URL with invalid scheme (not http/https) is rejected.""" + with pytest.raises(BadRequest, match="Only http and https are allowed"): + validate_destination_url("ftp://example.com") + + with pytest.raises(BadRequest, match="Only http and https are allowed"): + validate_destination_url("file:///etc/passwd") + + with pytest.raises(BadRequest, match="Only http and https are allowed"): + validate_destination_url("javascript:alert(1)") + + with pytest.raises(BadRequest, match="Only http and https are allowed"): + validate_destination_url("data:text/html,") + + def test_url_without_hostname_rejected(self) -> None: + """URL without hostname is rejected.""" + with pytest.raises(BadRequest, match="must include a hostname"): + validate_destination_url("http://") + + def test_localhost_hostname_rejected(self) -> None: + """Localhost hostname is rejected.""" + with pytest.raises(BadRequest, match="reserved loopback"): + validate_destination_url("http://localhost") + + with pytest.raises(BadRequest, match="reserved loopback"): + validate_destination_url("http://localhost:8080") + + with pytest.raises(BadRequest, match="reserved loopback"): + validate_destination_url("http://LOCALHOST") + + def test_valid_https_url_accepted(self) -> None: + """Valid HTTPS URL with public hostname is accepted.""" + # Should not raise - this is a valid public URL + validate_destination_url("https://www.example.com") + + def test_valid_http_url_accepted(self) -> None: + """Valid HTTP URL with public hostname is accepted.""" + # Should not raise + validate_destination_url("http://www.example.com/path") + + def test_url_with_port_accepted(self) -> None: + """Valid URL with port is accepted.""" + validate_destination_url("https://www.example.com:8443/path") + + +class TestValidateDestinationUrlIPv4Literal: + """IPv4 literal IP address validation tests.""" + + def test_loopback_ipv4_rejected(self) -> None: + """IPv4 loopback addresses are rejected.""" + with pytest.raises(BadRequest, match="loopback"): + validate_destination_url("http://127.0.0.1") + + with pytest.raises(BadRequest, match="loopback"): + validate_destination_url("http://127.0.0.2") + + with pytest.raises(BadRequest, match="loopback"): + validate_destination_url("http://127.255.255.255") + + def test_private_ipv4_rejected(self) -> None: + """IPv4 private addresses are rejected.""" + with pytest.raises(BadRequest, match="private"): + validate_destination_url("http://10.0.0.1") + + with pytest.raises(BadRequest, match="private"): + validate_destination_url("http://192.168.1.1") + + with pytest.raises(BadRequest, match="private"): + validate_destination_url("http://172.16.0.1") + + def test_link_local_ipv4_rejected(self) -> None: + """IPv4 link-local addresses are rejected.""" + with pytest.raises(BadRequest, match="link-local"): + validate_destination_url("http://169.254.1.1") + + def test_cloud_metadata_ipv4_rejected(self) -> None: + """Cloud metadata IP 169.254.169.254 is explicitly rejected.""" + with pytest.raises(BadRequest, match="cloud metadata"): + validate_destination_url("http://169.254.169.254") + + def test_multicast_ipv4_rejected(self) -> None: + """IPv4 multicast addresses are rejected.""" + with pytest.raises(BadRequest, match="multicast"): + validate_destination_url("http://224.0.0.1") + + with pytest.raises(BadRequest, match="multicast"): + validate_destination_url("http://239.255.255.255") + + def test_reserved_ipv4_rejected(self) -> None: + """IPv4 reserved addresses are rejected.""" + with pytest.raises(BadRequest, match="private|reserved"): + validate_destination_url("http://0.0.0.0") + + with pytest.raises(BadRequest, match="private|reserved"): + validate_destination_url("http://255.255.255.255") + + def test_public_ipv4_accepted(self) -> None: + """Public IPv4 addresses are accepted.""" + validate_destination_url("http://8.8.8.8") + validate_destination_url("http://1.1.1.1") + + +class TestValidateDestinationUrlIPv6Literal: + """IPv6 literal IP address validation tests.""" + + def test_ipv6_loopback_rejected(self) -> None: + """IPv6 loopback addresses are rejected.""" + with pytest.raises(BadRequest, match="loopback"): + validate_destination_url("http://[::1]") + + with pytest.raises(BadRequest, match="loopback"): + validate_destination_url("http://[0:0:0:0:0:0:0:1]") + + def test_ipv6_link_local_rejected(self) -> None: + """IPv6 link-local addresses are rejected.""" + with pytest.raises(BadRequest, match="link-local"): + validate_destination_url("http://[fe80::1]") + + def test_ipv6_private_rejected(self) -> None: + """IPv6 private addresses are rejected.""" + with pytest.raises(BadRequest, match="private"): + validate_destination_url("http://[fd00::1]") + + def test_ipv6_multicast_rejected(self) -> None: + """IPv6 multicast addresses are rejected.""" + with pytest.raises(BadRequest, match="multicast"): + validate_destination_url("http://[ff00::1]") + + def test_ipv6_public_accepted(self) -> None: + """Public IPv6 addresses are accepted.""" + validate_destination_url("http://[2001:4860:4860::8888]") + + +class TestValidateDestinationUrlHostnameResolution: + """Hostname resolution tests with various resolved IPs.""" + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_hostname_resolves_to_loopback_rejected( + self, mock_getaddrinfo: MagicMock + ) -> None: + """Hostname that resolves to loopback is rejected.""" + mock_getaddrinfo.return_value = [ + (2, 1, 6, "", ("127.0.0.1", 80)) + ] + with pytest.raises(BadRequest, match="loopback"): + validate_destination_url("http://internal.local") + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_hostname_resolves_to_private_rejected( + self, mock_getaddrinfo: MagicMock + ) -> None: + """Hostname that resolves to private IP is rejected.""" + mock_getaddrinfo.return_value = [ + (2, 1, 6, "", ("192.168.1.1", 80)) + ] + with pytest.raises(BadRequest, match="private"): + validate_destination_url("http://internal.local") + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_hostname_resolves_to_link_local_rejected( + self, mock_getaddrinfo: MagicMock + ) -> None: + """Hostname that resolves to link-local is rejected.""" + mock_getaddrinfo.return_value = [ + (2, 1, 6, "", ("169.254.1.1", 80)) + ] + with pytest.raises(BadRequest, match="link-local"): + validate_destination_url("http://internal.local") + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_hostname_resolves_to_cloud_metadata_rejected( + self, mock_getaddrinfo: MagicMock + ) -> None: + """Hostname that resolves to cloud metadata is rejected.""" + mock_getaddrinfo.return_value = [ + (2, 1, 6, "", ("169.254.169.254", 80)) + ] + with pytest.raises(BadRequest, match="cloud metadata"): + validate_destination_url("http://metadata.local") + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_hostname_resolves_to_multicast_rejected( + self, mock_getaddrinfo: MagicMock + ) -> None: + """Hostname that resolves to multicast is rejected.""" + mock_getaddrinfo.return_value = [ + (2, 1, 6, "", ("224.0.0.1", 80)) + ] + with pytest.raises(BadRequest, match="multicast"): + validate_destination_url("http://multicast.local") + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_hostname_resolves_to_reserved_rejected( + self, mock_getaddrinfo: MagicMock + ) -> None: + """Hostname that resolves to reserved IP is rejected.""" + mock_getaddrinfo.return_value = [ + (2, 1, 6, "", ("0.0.0.0", 80)) + ] + with pytest.raises(BadRequest, match="private|reserved"): + validate_destination_url("http://reserved.local") + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_hostname_resolves_to_public_ip_accepted( + self, mock_getaddrinfo: MagicMock + ) -> None: + """Hostname that resolves to public IP is accepted.""" + mock_getaddrinfo.return_value = [ + (2, 1, 6, "", ("8.8.8.8", 80)) + ] + validate_destination_url("http://example.com") + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_hostname_resolves_to_multiple_ips_rejects_if_any_private( + self, mock_getaddrinfo: MagicMock + ) -> None: + """If hostname resolves to multiple IPs and any is private, reject.""" + mock_getaddrinfo.return_value = [ + (2, 1, 6, "", ("8.8.8.8", 80)), + (2, 1, 6, "", ("192.168.1.1", 80)), + ] + with pytest.raises(BadRequest, match="private"): + validate_destination_url("http://example.com") + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_hostname_resolution_returns_empty_list_allowed( + self, mock_getaddrinfo: MagicMock + ) -> None: + """Empty resolution result (no IPs found) is allowed for shortener.""" + mock_getaddrinfo.return_value = [] + # Should not raise - shortener doesn't fetch, just redirects + validate_destination_url("http://nonexistent.example.com") + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_hostname_resolution_failure_allowed( + self, mock_getaddrinfo: MagicMock + ) -> None: + """Hostname resolution failure (socket.gaierror) is allowed.""" + import socket as socket_module + + mock_getaddrinfo.side_effect = socket_module.gaierror("NXDOMAIN") + # Should not raise - shortener doesn't fetch + validate_destination_url("http://nonexistent.example.local") + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_hostname_resolution_socket_error_rejected( + self, mock_getaddrinfo: MagicMock + ) -> None: + """Generic socket error during resolution is rejected.""" + import socket as socket_module + + mock_getaddrinfo.side_effect = socket_module.error("Network unreachable") + with pytest.raises(BadRequest, match="Error validating"): + validate_destination_url("http://example.com") + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_invalid_ip_in_resolution_result_rejected( + self, mock_getaddrinfo: MagicMock + ) -> None: + """Invalid IP address in resolution result is rejected.""" + mock_getaddrinfo.return_value = [ + (2, 1, 6, "", ("not-a-valid-ip", 80)) + ] + with pytest.raises(BadRequest, match="Invalid IP address"): + validate_destination_url("http://example.com") + + @patch("app.urlvalidation.socket.getaddrinfo") + def test_ipv6_resolution_accepted(self, mock_getaddrinfo: MagicMock) -> None: + """IPv6 resolution from hostname accepted if public.""" + # IPv6 return format: (family, socktype, proto, canonname, (host, port, flowinfo, scopeid)) + mock_getaddrinfo.return_value = [ + (10, 1, 6, "", ("2001:4860:4860::8888", 80, 0, 0)) + ] + validate_destination_url("http://example.com") + + +class TestValidateDestinationUrlEdgeCases: + """Edge case tests for URL validation.""" + + def test_url_parse_error_rejected(self) -> None: + """URL that fails to parse is rejected.""" + # Very pathological URL that might fail parsing + # (most URLs parse fine, but we still need to cover the exception path) + # Actually, urlparse is very permissive, so let's just ensure the exception + # is caught if it happens + with patch("app.urlvalidation.urlparse") as mock_parse: + mock_parse.side_effect = Exception("Parse error") + with pytest.raises(BadRequest, match="Invalid URL"): + validate_destination_url("http://example.com") + + def test_scheme_case_insensitive(self) -> None: + """URL scheme matching is case-insensitive.""" + validate_destination_url("HTTPS://www.example.com") + validate_destination_url("HTTP://www.example.com") + validate_destination_url("HtTpS://www.example.com") + + +# --------------------------------------------------------------------------- +# validate_short_code Tests +# --------------------------------------------------------------------------- + + +class TestValidateShortCodeBasic: + """Basic short code validation tests.""" + + def test_empty_code_rejected(self) -> None: + """Empty short code is rejected.""" + with pytest.raises(BadRequest, match="cannot be empty"): + validate_short_code("") + + def test_none_code_rejected(self) -> None: + """None short code is rejected.""" + with pytest.raises(BadRequest, match="cannot be empty"): + validate_short_code(None) # type: ignore + + def test_alphanumeric_code_accepted(self) -> None: + """Alphanumeric codes are accepted.""" + validate_short_code("abc123") + validate_short_code("ABC123") + validate_short_code("aBc123") + + def test_code_with_hyphens_accepted(self) -> None: + """Codes with hyphens are accepted.""" + validate_short_code("my-link") + validate_short_code("a-b-c") + + def test_code_with_underscores_accepted(self) -> None: + """Codes with underscores are accepted.""" + validate_short_code("my_link") + validate_short_code("a_b_c") + + def test_code_with_mixed_allowed_chars_accepted(self) -> None: + """Codes with mix of letters, digits, hyphens, underscores accepted.""" + validate_short_code("my-code_123") + validate_short_code("a1b2c3-d_e") + + def test_code_too_long_rejected(self) -> None: + """Short codes longer than 32 chars are rejected.""" + long_code = "a" * 33 + with pytest.raises(BadRequest, match="32 characters or less"): + validate_short_code(long_code) + + def test_code_exactly_32_chars_accepted(self) -> None: + """Short codes exactly 32 chars are accepted.""" + code_32 = "a" * 32 + validate_short_code(code_32) + + def test_code_with_space_rejected(self) -> None: + """Codes with spaces are rejected.""" + with pytest.raises(BadRequest, match="alphanumeric characters, hyphens, and underscores"): + validate_short_code("my link") + + def test_code_with_special_chars_rejected(self) -> None: + """Codes with special characters are rejected.""" + with pytest.raises(BadRequest, match="alphanumeric characters, hyphens, and underscores"): + validate_short_code("my@link") + + with pytest.raises(BadRequest, match="alphanumeric characters, hyphens, and underscores"): + validate_short_code("my.link") + + with pytest.raises(BadRequest, match="alphanumeric characters, hyphens, and underscores"): + validate_short_code("my/link") + + with pytest.raises(BadRequest, match="alphanumeric characters, hyphens, and underscores"): + validate_short_code("my!link") + + +class TestValidateShortCodeReserved: + """Reserved path validation tests.""" + + def test_reserved_paths_rejected(self) -> None: + """Reserved paths cannot be used as short codes.""" + reserved = [ + "api", "admin", "health", "healthz", "readyz", "login", "logout", + "auth", "static", "assets", "urls", "collections", "settings", + "users", "teams", "tenants", "roles", "docs", "swagger", "openapi", + ] + for path in reserved: + with pytest.raises(BadRequest, match="reserved"): + validate_short_code(path) + + def test_reserved_paths_case_insensitive(self) -> None: + """Reserved path checking is case-insensitive.""" + with pytest.raises(BadRequest, match="reserved"): + validate_short_code("API") + + with pytest.raises(BadRequest, match="reserved"): + validate_short_code("Admin") + + with pytest.raises(BadRequest, match="reserved"): + validate_short_code("HEALTH") + + def test_non_reserved_codes_accepted(self) -> None: + """Non-reserved codes are accepted.""" + validate_short_code("mylink") + validate_short_code("test123") + validate_short_code("my-short-code")