Add anonymous URL ownership claim via manage_token

repositories/url_repository.py

@@ -14,6 +14,7 @@
 from pymongo.asynchronous.collection import AsyncCollection
 from pymongo.errors import DuplicateKeyError, PyMongoError
 
+from schemas.models.base import ANONYMOUS_OWNER_ID
 from schemas.models.url import UrlV2Doc
 from shared.logging import get_logger
 
@@ -235,3 +236,39 @@ async def check_stats_privacy(self, alias: str) -> dict:
                 error_type=type(exc).__name__,
             )
             raise
+    async def claim_by_manage_token(
+        self,
+        alias: str,
+        token_hash: str,
+        new_owner_id: ObjectId,
+    ) -> bool:
+        """
+        Atomically claim an anonymous URL if the token hash matches and it has no owner.
+
+        Returns True if the claim succeeded (document was modified), False otherwise.
+        """
+        try:
+            result = await self._col.update_one(
+                {
+                    "alias": alias,
+                    "owner_id": ANONYMOUS_OWNER_ID,
+                    "manage_token": token_hash,
+                },
+                {
+                    "$set": {
+                        "owner_id": new_owner_id,
+                        "manage_token": None,
+                        "updated_at": datetime.now(timezone.utc),
+                    }
+                },
+            )
+            return result.modified_count > 0
+        except PyMongoError as exc:
+            log.error(
+                "url_repo_claim_by_manage_token_failed",
+                alias=alias,
+                error=str(exc),
+                error_type=type(exc).__name__,
+            )
+            raise
+

routes/api_v1/__init__.py

@@ -2,10 +2,11 @@
 
 from fastapi import APIRouter
 
-from routes.api_v1 import exports, keys, management, shorten, stats, urls
+from routes.api_v1 import claim, exports, keys, management, shorten, stats, urls
 
 router = APIRouter(prefix="/api/v1")
 router.include_router(shorten.router)
+router.include_router(claim.router)
 router.include_router(urls.router)
 router.include_router(management.router)
 router.include_router(stats.router)

routes/api_v1/claim.py

@@ -0,0 +1,63 @@
+"""
+POST /api/v1/claim — claim ownership of an anonymously created URL.
+
+Transfers a URL from anonymous ownership to the authenticated user,
+consuming and invalidating the one-time manage_token in the process.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from pydantic import BaseModel, Field
+
+from dependencies import CurrentUser, get_url_service, require_auth
+from middleware.rate_limiter import Limits, dynamic_limit, limiter
+from services.url_service import UrlService
+
+router = APIRouter(tags=["URL Management"])
+
+_claim_limit, _claim_key = dynamic_limit(Limits.API_AUTHED, Limits.API_AUTHED)
+
+
+class ClaimUrlRequest(BaseModel):
+    alias: str = Field(..., description="The short alias to claim.")
+    manage_token: str = Field(..., description="The one-time token returned at creation.")
+
+
+class ClaimUrlResponse(BaseModel):
+    success: bool
+    message: str
+
+
+@router.post(
+    "/claim",
+    status_code=200,
+    operation_id="claimUrl",
+    summary="Claim Anonymous URL",
+)
+@limiter.limit(_claim_limit, key_func=_claim_key)
+async def claim_url(
+    request: Request,
+    body: ClaimUrlRequest,
+    user: CurrentUser = Depends(require_auth),
+    url_service: UrlService = Depends(get_url_service),
+) -> ClaimUrlResponse:
+    """
+    Transfer ownership of an anonymously created URL to your account.
+
+    The manage_token is single-use and is invalidated immediately on success.
+    Returns 403 if the token is wrong, 409 if already claimed, 404 if not found.
+    """
+    claimed = await url_service.claim_url(
+        alias=body.alias,
+        raw_token=body.manage_token,
+        new_owner_id=user.user_id,
+    )
+    if not claimed:
+        # Don't distinguish between wrong token / already claimed / not found
+        # to avoid oracle attacks.
+        raise HTTPException(
+            status_code=403,
+            detail="Invalid token or URL is not claimable.",
+        )
+    return ClaimUrlResponse(success=True, message="URL successfully claimed.")

routes/api_v1/shorten.py

@@ -69,7 +69,7 @@ async def shorten_v1(
     owner_id = user.user_id if user is not None else None
     client_ip = get_client_ip(request)
 
-    doc = await url_service.create(body, owner_id, client_ip)
+    doc, raw_token = await url_service.create(body, owner_id, client_ip)
 
     settings = request.app.state.settings
     return UrlResponse(
@@ -80,4 +80,5 @@ async def shorten_v1(
         created_at=to_unix_timestamp(doc.created_at, default=0),
         status=doc.status,
         private_stats=doc.private_stats,
+        manage_token=raw_token,
     )

schemas/dto/responses/url.py

@@ -49,6 +49,12 @@ class UrlResponse(BaseModel):
         default=None,
         description="Whether statistics are private (owner-only).",
     )
+    manage_token: Optional[str] = Field(
+        default=None,
+        description="Returned once for anonymous URLs only. Never stored in plaintext.",
+        examples=["kS9_x1z..."],
+    )
+
 
 
 class UpdateUrlResponse(BaseModel):

schemas/models/url.py

@@ -49,6 +49,8 @@ def _coerce_null_owner(cls, v: Any) -> Any:
     total_clicks: int = 0
     last_click: Optional[datetime] = None
     updated_at: Optional[datetime] = None
+    manage_token: Optional[str] = None
+
 
 
 class LegacyUrlDoc(MongoBaseModel):

services/url_service.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import re
+import secrets
 import time
 from datetime import datetime, timezone
 from typing import Optional
@@ -39,7 +40,7 @@
 from schemas.dto.requests.url import CreateUrlRequest, ListUrlsQuery, UpdateUrlRequest
 from schemas.models.base import ANONYMOUS_OWNER_ID
 from schemas.models.url import EmojiUrlDoc, LegacyUrlDoc, UrlV2Doc
-from shared.crypto import hash_password
+from shared.crypto import hash_password, hash_token
 from shared.datetime_utils import parse_datetime
 from shared.generators import generate_short_code_v2
 from shared.logging import get_logger, should_sample
@@ -165,7 +166,7 @@ async def create(
         request: CreateUrlRequest,
         owner_id: Optional[ObjectId],
         client_ip: str,
-    ) -> UrlV2Doc:
+    ) -> tuple[UrlV2Doc, Optional[str]]:
         """
         Create a new shortened URL.
 
@@ -252,6 +253,13 @@ async def create(
             "total_clicks": 0,
             "last_click": None,
         }
+
+        raw_token: Optional[str] = None
+        if owner_id is None:
+            raw_token = secrets.token_urlsafe(32)
+            doc["manage_token"] = hash_token(raw_token)
+        else:
+            doc["manage_token"] = None
 
         # 8. Insert
         inserted_id = await self._url_repo.insert(doc)
@@ -270,7 +278,7 @@ async def create(
             private_stats=private_stats,
         )
 
-        return UrlV2Doc.from_mongo(doc)
+        return UrlV2Doc.from_mongo(doc), raw_token
 
     async def update(
         self,
@@ -507,6 +515,11 @@ async def list_by_owner(
             "sortOrder": "descending" if sort_order == -1 else "ascending",
         }
 
+    async def claim_url(self, alias: str, raw_token: str, new_owner_id: ObjectId) -> bool:
+        """Transfer ownership of an anonymous URL to an authenticated user."""
+        token_hash = hash_token(raw_token)
+        return await self._url_repo.claim_by_manage_token(alias, token_hash, new_owner_id)
+
     # ── Private helpers ───────────────────────────────────────────────────────
 
     async def _dispatch(self, short_code: str) -> tuple[Optional[UrlCacheData], str]:

static/css/index.css

@@ -623,4 +623,45 @@ label.cbx {
         flex-direction: column;
         gap: 10px;
     }
+}
+
+/* ── Anonymous URL claim UI ───────────────────────────────── */
+
+.unclaimed-badge {
+    display: inline-block;
+    font-size: 10px;
+    font-weight: 600;
+    letter-spacing: 0.5px;
+    color: rgba(255, 200, 60, 0.9);
+    background: rgba(255, 200, 60, 0.1);
+    border: 1px solid rgba(255, 200, 60, 0.3);
+    border-radius: 4px;
+    padding: 2px 6px;
+    margin-left: 8px;
+    vertical-align: middle;
+}
+
+#claim-all-wrapper {
+    display: flex;
+    justify-content: center;
+    margin-top: 10px;
+    padding: 0 10px;
+}
+
+#claim-all-btn {
+    background: rgba(255, 200, 60, 0.1);
+    border: 1px solid rgba(255, 200, 60, 0.35);
+    color: rgba(255, 220, 100, 0.95);
+    padding: 9px 20px;
+    border-radius: 8px;
+    cursor: pointer;
+    font-size: 13px;
+    font-family: inherit;
+    transition: background 0.2s ease;
+    width: 100%;
+    max-width: 340px;
+}
+
+#claim-all-btn:hover {
+    background: rgba(255, 200, 60, 0.2);
 }