Implement v1 URL shortener (Bearer API, public 302, landing, CI) (#2)
Deploy Beta / unit-tests (push) Successful in 4s
Deploy Beta / docker (push) Successful in 13s
Deploy Beta / deploy-beta (push) Successful in 2m38s

## Summary

- Standalone Django 6 shortener: Bearer `/api/links/` (create/list/detail/disable) and public `GET /<code>` 302
- Host split, target-host allowlist, named rotatable tokens; `short_url` from `PUBLIC_SHORT_URL`
- Landing page, DEBUG-only `/debug/` mint form, Django admin
- Docker/compose (host **8005**), Gitea CI like monica_site (PR tests, beta on merge, prod button)
- Caller contract in `API.md`

Closes #1. Infra follow-up: [server-infra#22](ai_ml_operations/server-infra#22).

## Test plan
- [ ] `cd site && uv run python manage.py test`
- [ ] `docker compose up --build` → http://127.0.0.1:8005/
- [ ] `POST /api/links/` with `Bearer monica:dev-only-token` → 201
- [ ] `GET /<code>` → 302 to allowlisted https URL
- [ ] No Bearer → 401; non-allowlisted host → 400
- [ ] `/debug/` only when `DEBUG=true`

Reviewed-on: #2
This commit was merged in pull request #2.
This commit is contained in:
2026-08-30 04:55:33 -07:00
parent 4baaa4b33c
commit 630b770c12
48 changed files with 3469 additions and 2 deletions
+179
View File
@@ -0,0 +1,179 @@
"""Link minting, target-URL allowlist, click recording."""
from __future__ import annotations
import hashlib
import hmac
import logging
import secrets
from datetime import datetime
from urllib.parse import urlsplit, urlunsplit
from django.conf import settings
from django.db import IntegrityError
from django.db.models import F
from django.http import HttpRequest
from django.utils import timezone
from django.utils.dateparse import parse_datetime
from links.models import Click, ShortLink
logger = logging.getLogger(__name__)
MAX_CODE_ATTEMPTS = 8
class ValidationError(ValueError):
pass
class CodeCollisionError(RuntimeError):
pass
def host_allowed(hostname: str, allowed: list[str]) -> bool:
hostname = hostname.lower().rstrip(".")
for entry in allowed:
entry = entry.lower().strip()
if entry.startswith("*."):
entry = entry[2:]
entry = entry.lstrip(".").rstrip(".")
if not entry:
continue
if hostname == entry or hostname.endswith("." + entry):
return True
return False
def validate_target_url(raw: str) -> str:
"""Return a canonical https URL or raise ValidationError."""
if not raw or not isinstance(raw, str):
raise ValidationError("invalid url")
raw = raw.strip()
if raw.startswith("//"):
raise ValidationError("invalid url")
try:
parts = urlsplit(raw)
except ValueError as exc:
raise ValidationError("invalid url") from exc
if parts.scheme.lower() != "https":
raise ValidationError("invalid url")
if parts.username or parts.password:
raise ValidationError("invalid url")
hostname = (parts.hostname or "").lower().rstrip(".")
if not hostname:
raise ValidationError("invalid url")
allowed = list(getattr(settings, "SHORT_ALLOWED_HOSTS", []) or [])
if not host_allowed(hostname, allowed):
raise ValidationError("host not allowlisted")
netloc = hostname
if parts.port:
netloc = f"{hostname}:{parts.port}"
return urlunsplit(("https", netloc, parts.path, parts.query, parts.fragment))
def generate_code(length: int | None = None) -> str:
alphabet = settings.CODE_ALPHABET
size = length if length is not None else settings.SHORT_CODE_LENGTH
return "".join(secrets.choice(alphabet) for _ in range(size))
def mint_unique_code() -> str:
for _ in range(MAX_CODE_ATTEMPTS):
code = generate_code()
if not ShortLink.objects.filter(code=code).exists():
return code
raise CodeCollisionError("could not allocate a unique code")
def parse_expires_at(value) -> datetime | None:
if value in (None, ""):
return None
if not isinstance(value, str):
raise ValidationError("invalid expires_at")
parsed = parse_datetime(value)
if parsed is None:
raise ValidationError("invalid expires_at")
if timezone.is_naive(parsed):
parsed = timezone.make_aware(parsed, timezone.get_current_timezone())
return parsed
def find_idempotent_link(target_url: str, external_ref: str) -> ShortLink | None:
if not external_ref:
return None
qs = ShortLink.objects.filter(
target_url=target_url,
external_ref=external_ref,
is_active=True,
)
now = timezone.now()
for link in qs:
if link.expires_at is None or link.expires_at > now:
return link
return None
def create_link(
*,
target_url: str,
title: str,
external_ref: str,
expires_at: datetime | None,
token_name: str,
) -> tuple[ShortLink, bool]:
"""Return ``(link, created)``. ``created`` is False on idempotent hit."""
existing = find_idempotent_link(target_url, external_ref)
if existing:
return existing, False
for _ in range(MAX_CODE_ATTEMPTS):
try:
link = ShortLink.objects.create(
code=generate_code(),
target_url=target_url,
title=title,
external_ref=external_ref,
expires_at=expires_at,
created_by_token=token_name,
)
return link, True
except IntegrityError:
continue
raise CodeCollisionError("could not allocate a unique code")
def client_ip(request: HttpRequest) -> str:
forwarded = request.META.get("HTTP_X_FORWARDED_FOR") or ""
if forwarded:
return forwarded.split(",")[0].strip()
return (request.META.get("REMOTE_ADDR") or "").strip()
def hash_ip(ip: str) -> str:
pepper = getattr(settings, "CLICK_IP_PEPPER", "") or ""
if not ip or not pepper:
return ""
return hmac.new(pepper.encode(), ip.encode(), hashlib.sha256).hexdigest()
def record_click(request: HttpRequest, link: ShortLink) -> None:
Click.objects.create(
link=link,
ip_hash=hash_ip(client_ip(request)),
user_agent=(request.META.get("HTTP_USER_AGENT") or "")[:512],
referrer=(request.META.get("HTTP_REFERER") or "")[:1024],
)
ShortLink.objects.filter(pk=link.pk).update(click_count=F("click_count") + 1)
def record_click_best_effort(request: HttpRequest, link: ShortLink) -> None:
try:
record_click(request, link)
except Exception:
logger.exception("click record failed for code=%s", link.code)