ai_ml_operations/url_shortening_service
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add v1 shortener: Bearer API, public 302, landing, and CI.
CI / test (pull_request) Successful in 6s
Details

Browse Source 
Standalone Django service so callers can mint links and phones get a 302.
Closes #1.
This commit is contained in:
westfarn
2026-08-30 06:54:32 -05:00
parent 4baaa4b33c
commit 143ac7c6d0
48 changed files with 3469 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
site/core/__init__.py
+1
View File
@@ -0,0 +1 @@
site/core/apps.py
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class CoreConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "core"
site/core/context_processors.py
+14
View File
@@ -0,0 +1,14 @@
from django.conf import settings
from links.middleware import is_public_host
def branding(request):
return {
"SITE_NAME": settings.SITE_NAME,
"SHORT_DOMAIN": settings.SHORT_DOMAIN,
"PUBLIC_SHORT_URL": settings.PUBLIC_SHORT_URL,
"CREDIT_NAME": settings.CREDIT_NAME,
"CREDIT_URL": settings.CREDIT_URL,
"show_debug_create": settings.DEBUG and not is_public_host(request.get_host()),
}
site/core/templates/core/base.html
+128
View File
@@ -0,0 +1,128 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{% block title %}{{ SITE_NAME }}{% endblock %}</title>
<style>
:root {
--ink: #1c1914;
--paper: #f3ead8;
--rule: #c9b896;
--pine: #2a4a3c;
--rust: #9a3f24;
--muted: #6b6254;
}
* { box-sizing: border-box; }
html, body { margin: 0; min-height: 100%; }
body {
background: var(--paper);
color: var(--ink);
font-family: "Iowan Old Style", Palatino, "Palatino Linotype", "Times New Roman", serif;
line-height: 1.5;
}
.shell {
max-width: 40rem;
margin: 0 auto;
padding: 3.5rem 1.5rem 2rem;
}
.eyebrow {
font-family: ui-monospace, "Cascadia Code", "SF Mono", Menlo, monospace;
font-size: 0.72rem;
letter-spacing: 0.16em;
text-transform: uppercase;
color: var(--pine);
margin: 0 0 0.75rem;
}
h1 {
font-size: clamp(2.4rem, 8vw, 4.2rem);
font-weight: 700;
letter-spacing: -0.03em;
line-height: 0.95;
margin: 0 0 1.25rem;
}
p { margin: 0 0 1rem; font-size: 1.15rem; }
.lede { color: var(--muted); max-width: 32rem; }
a { color: var(--rust); }
a:hover { color: var(--ink); }
footer {
margin-top: 3.5rem;
padding-top: 1rem;
border-top: 1px solid var(--rule);
font-family: ui-monospace, "Cascadia Code", "SF Mono", Menlo, monospace;
font-size: 0.78rem;
color: var(--muted);
}
footer a { color: inherit; }
.banner {
font-family: ui-monospace, Menlo, monospace;
font-size: 0.75rem;
letter-spacing: 0.08em;
text-transform: uppercase;
background: var(--ink);
color: var(--paper);
padding: 0.4rem 1.5rem;
}
form { margin-top: 1.75rem; display: grid; gap: 0.9rem; }
label {
display: grid;
gap: 0.3rem;
font-family: ui-monospace, Menlo, monospace;
font-size: 0.78rem;
letter-spacing: 0.04em;
text-transform: uppercase;
color: var(--pine);
}
input[type="url"], input[type="text"] {
font: 1rem/1.4 "Iowan Old Style", Palatino, serif;
padding: 0.55rem 0.65rem;
border: 1px solid var(--rule);
background: #fffdf6;
color: var(--ink);
}
button, .btn {
appearance: none;
border: 0;
background: var(--pine);
color: var(--paper);
font-family: ui-monospace, Menlo, monospace;
font-size: 0.8rem;
letter-spacing: 0.08em;
text-transform: uppercase;
padding: 0.7rem 1rem;
cursor: pointer;
text-decoration: none;
display: inline-block;
width: fit-content;
}
button:hover, .btn:hover { background: var(--ink); }
.error { color: var(--rust); font-size: 0.95rem; }
.result {
margin-top: 1.5rem;
padding: 1rem;
border: 1px dashed var(--pine);
background: #fffdf6;
}
.result code {
font-size: 1.05rem;
word-break: break-all;
}
.messages { list-style: none; padding: 0; margin: 0 0 1rem; color: var(--pine); }
</style>
</head>
<body>
{% if debug %}
<div class="banner">Debug — create page is local only</div>
{% endif %}
<main class="shell">
{% block content %}{% endblock %}
<footer>
Created by
<a href="{{ CREDIT_URL }}">{{ CREDIT_NAME }}</a>
{% if show_debug_create %}
· <a href="{% url 'debug-create' %}">Create a short link</a>
{% endif %}
</footer>
</main>
</body>
</html>
site/core/templates/core/landing.html
+14
View File
@@ -0,0 +1,14 @@
{% extends "core/base.html" %}
{% block title %}{{ SITE_NAME }}{% endblock %}
{% block content %}
<p class="eyebrow">{{ SITE_NAME }}</p>
<h1>{{ SHORT_DOMAIN }}</h1>
<p class="lede">
This is a URL shortening service. A path like
<strong>/a3k9xm</strong> sends you to the long HTTPS URL.
</p>
<p class="lede">
Created by <a href="{{ CREDIT_URL }}">{{ CREDIT_NAME }}</a>
· <a href="{{ CREDIT_URL }}">aimloperations.com</a>
</p>
{% endblock %}
site/core/views.py
+12
View File
@@ -0,0 +1,12 @@
from django.http import JsonResponse
from django.shortcuts import render
def healthz(_request):
"""Liveness probe for deploy / NPM health checks."""
return JsonResponse({"status": "ok"})
def landing(request):
"""Public root: this is a URL shortening service, not an API index."""
return render(request, "core/landing.html")
site/links/__init__.py
+1
View File
@@ -0,0 +1 @@
site/links/admin.py
+158
View File
@@ -0,0 +1,158 @@
from django.contrib import admin, messages
from django.forms import ModelForm, ValidationError as FormValidationError
from django.utils.html import format_html
from links.models import Click, ShortLink
from links.services import ValidationError, generate_code, validate_target_url
admin.site.site_header = "URL shortener"
admin.site.site_title = "Shortener admin"
admin.site.index_title = "Links and clicks"
class ShortLinkAdminForm(ModelForm):
class Meta:
model = ShortLink
fields = "__all__"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if "code" in self.fields:
self.fields["code"].required = False
if "created_by_token" in self.fields:
self.fields["created_by_token"].required = False
def clean_target_url(self):
raw = self.cleaned_data.get("target_url") or ""
try:
return validate_target_url(raw)
except ValidationError as exc:
raise FormValidationError(str(exc)) from exc
def clean(self):
cleaned = super().clean()
if not self.instance.pk and not cleaned.get("code"):
cleaned["code"] = generate_code()
if not cleaned.get("created_by_token"):
cleaned["created_by_token"] = "admin"
return cleaned
class ClickInline(admin.TabularInline):
model = Click
extra = 0
can_delete = False
show_change_link = True
max_num = 0
readonly_fields = ("occurred_at", "ip_hash", "user_agent", "referrer")
fields = ("occurred_at", "ip_hash", "user_agent", "referrer")
def has_add_permission(self, request, obj=None):
return False
@admin.register(ShortLink)
class ShortLinkAdmin(admin.ModelAdmin):
form = ShortLinkAdminForm
list_display = (
"code",
"short_url_display",
"target_url",
"title",
"is_active",
"click_count",
"created_by_token",
"expires_at",
"created_at",
)
list_filter = ("is_active", "created_by_token", "created_at")
search_fields = ("code", "target_url", "external_ref", "title")
date_hierarchy = "created_at"
ordering = ("-created_at",)
list_per_page = 50
actions = ("disable_links",)
inlines = (ClickInline,)
readonly_fields = (
"id",
"code",
"created_by_token",
"click_count",
"created_at",
"updated_at",
"short_url_display",
)
fieldsets = (
(
None,
{
"fields": (
"code",
"short_url_display",
"target_url",
"title",
"external_ref",
"is_active",
"expires_at",
)
},
),
(
"Attribution",
{"fields": ("created_by_token", "click_count", "created_at", "updated_at", "id")},
),
)
def get_readonly_fields(self, request, obj=None):
if obj:
return self.readonly_fields + ("target_url",)
return self.readonly_fields
@admin.display(description="Short URL")
def short_url_display(self, obj: ShortLink) -> str:
if not obj.code:
return "—"
return format_html('<a href="{0}">{0}</a>', obj.public_short_url)
def save_model(self, request, obj, form, change):
if not change:
if not obj.code:
obj.code = generate_code()
if not obj.created_by_token:
obj.created_by_token = (
request.user.get_username() if request.user.is_authenticated else "admin"
)
super().save_model(request, obj, form, change)
@admin.action(description="Disable selected links")
def disable_links(self, request, queryset):
updated = queryset.filter(is_active=True).update(is_active=False)
self.message_user(
request,
f"Disabled {updated} link(s).",
messages.SUCCESS if updated else messages.INFO,
)
@admin.register(Click)
class ClickAdmin(admin.ModelAdmin):
list_display = ("link", "occurred_at", "ip_hash_short", "user_agent_short")
list_filter = ("occurred_at",)
search_fields = ("link__code", "ip_hash", "user_agent", "referrer")
date_hierarchy = "occurred_at"
readonly_fields = ("id", "link", "occurred_at", "ip_hash", "user_agent", "referrer")
ordering = ("-occurred_at",)
def has_add_permission(self, request):
return False
def has_change_permission(self, request, obj=None):
return False
@admin.display(description="IP hash")
def ip_hash_short(self, obj: Click) -> str:
return (obj.ip_hash[:12] + "…") if obj.ip_hash else "—"
@admin.display(description="User agent")
def user_agent_short(self, obj: Click) -> str:
ua = obj.user_agent or ""
return (ua[:48] + "…") if len(ua) > 48 else (ua or "—")
site/links/api.py
+156
View File
@@ -0,0 +1,156 @@
"""Internal JSON API for minting and managing short links."""
from __future__ import annotations
import json
from django.http import Http404, HttpRequest, JsonResponse
from django.views.decorators.http import require_GET, require_http_methods, require_POST
from links.auth import require_bearer
from links.models import ShortLink
from links.services import (
CodeCollisionError,
ValidationError,
create_link,
parse_expires_at,
validate_target_url,
)
def _json_body(request: HttpRequest) -> dict | None:
if not request.body:
return {}
try:
data = json.loads(request.body)
except json.JSONDecodeError:
return None
if not isinstance(data, dict):
return None
return data
def _iso(dt) -> str | None:
if dt is None:
return None
return dt.isoformat().replace("+00:00", "Z")
def serialize_link(link: ShortLink) -> dict:
from django.conf import settings
origin = (settings.PUBLIC_SHORT_URL or "").rstrip("/")
return {
"code": link.code,
"short_url": f"{origin}/{link.code}",
"target_url": link.target_url,
"title": link.title,
"is_active": link.is_active,
"click_count": link.click_count,
"created_at": _iso(link.created_at),
}
@require_bearer
@require_http_methods(["GET", "POST"])
def links_collection(request: HttpRequest):
if request.method == "POST":
return _create(request)
return _list(request)
def _create(request: HttpRequest) -> JsonResponse:
data = _json_body(request)
if data is None:
return JsonResponse({"detail": "invalid json"}, status=400)
raw_url = data.get("target_url")
try:
target_url = validate_target_url(raw_url if isinstance(raw_url, str) else "")
expires_at = parse_expires_at(data.get("expires_at"))
except ValidationError as exc:
return JsonResponse({"detail": str(exc)}, status=400)
title = data.get("title") or ""
if not isinstance(title, str):
return JsonResponse({"detail": "invalid title"}, status=400)
title = title[:200]
external_ref = data.get("external_ref") or ""
if not isinstance(external_ref, str):
return JsonResponse({"detail": "invalid external_ref"}, status=400)
external_ref = external_ref[:64]
try:
link, created = create_link(
target_url=target_url,
title=title,
external_ref=external_ref,
expires_at=expires_at,
token_name=request.token_name,
)
except CodeCollisionError:
return JsonResponse({"detail": "could not allocate a unique code"}, status=500)
return JsonResponse(serialize_link(link), status=201 if created else 200)
def _list(request: HttpRequest) -> JsonResponse:
qs = ShortLink.objects.all()
external_ref = request.GET.get("external_ref")
if external_ref is not None:
qs = qs.filter(external_ref=external_ref)
is_active = request.GET.get("is_active")
if is_active is not None:
lowered = is_active.lower()
if lowered in {"true", "1"}:
qs = qs.filter(is_active=True)
elif lowered in {"false", "0"}:
qs = qs.filter(is_active=False)
else:
return JsonResponse({"detail": "invalid is_active"}, status=400)
try:
limit = int(request.GET.get("limit", 20))
offset = int(request.GET.get("offset", 0))
except (TypeError, ValueError):
return JsonResponse({"detail": "invalid pagination"}, status=400)
limit = min(max(limit, 0), 100)
offset = max(offset, 0)
total = qs.count()
rows = list(qs[offset : offset + limit])
return JsonResponse(
{
"count": total,
"limit": limit,
"offset": offset,
"results": [serialize_link(link) for link in rows],
}
)
@require_bearer
@require_GET
def link_detail(request: HttpRequest, code: str):
try:
link = ShortLink.objects.get(code=code)
except ShortLink.DoesNotExist as exc:
raise Http404() from exc
return JsonResponse(serialize_link(link))
@require_bearer
@require_POST
def link_disable(request: HttpRequest, code: str):
try:
link = ShortLink.objects.get(code=code)
except ShortLink.DoesNotExist as exc:
raise Http404() from exc
if link.is_active:
link.is_active = False
link.save(update_fields=["is_active", "updated_at"])
return JsonResponse(serialize_link(link))
site/links/apps.py
+6
View File
@@ -0,0 +1,6 @@
from django.apps import AppConfig
class LinksConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "links"
site/links/auth.py
+64
View File
@@ -0,0 +1,64 @@
"""Bearer token auth for /api/. The lock that keeps a public API host closed."""
from __future__ import annotations
import hmac
from collections.abc import Callable
from functools import wraps
from django.conf import settings
from django.http import HttpRequest, JsonResponse
from django.views.decorators.csrf import csrf_exempt
def parse_bearer(request: HttpRequest) -> str | None:
header = request.META.get("HTTP_AUTHORIZATION") or ""
if not header.startswith("Bearer "):
return None
token = header[7:].strip()
return token or None
def authenticate_token(raw_token: str | None) -> str | None:
"""Return the configured token name, or None if no match.
Accepts either ``name:secret`` (as callers send) or the bare secret.
Always compares against every configured token (constant-time).
"""
configured = list(getattr(settings, "SHORTENER_API_TOKENS", []) or [])
if not configured or not raw_token:
return None
matched_name: str | None = None
for name, secret in configured:
full = f"{name}:{secret}"
if hmac.compare_digest(raw_token, full) or hmac.compare_digest(raw_token, secret):
matched_name = name
# Keep looping so compare_digest runs for every token.
return matched_name
def token_name_for_request(request: HttpRequest) -> str | None:
return authenticate_token(parse_bearer(request))
def require_bearer(view: Callable) -> Callable:
"""Decorator: 503 if no tokens configured, 401 if missing/wrong Bearer."""
@csrf_exempt
@wraps(view)
def wrapper(request, *args, **kwargs):
configured = list(getattr(settings, "SHORTENER_API_TOKENS", []) or [])
if not configured:
return JsonResponse({"detail": "Service unavailable"}, status=503)
name = token_name_for_request(request)
if not name:
response = JsonResponse({"detail": "Unauthorized"}, status=401)
response["WWW-Authenticate"] = "Bearer"
return response
request.token_name = name
return view(request, *args, **kwargs)
return wrapper
site/links/debug_views.py
+41
View File
@@ -0,0 +1,41 @@
"""DEBUG-only form to mint a short link without curling the API."""
from django.conf import settings
from django.contrib import messages
from django.http import Http404
from django.shortcuts import render
from django.views.decorators.http import require_http_methods
from links.forms import DebugCreateForm
from links.services import CodeCollisionError, create_link
@require_http_methods(["GET", "POST"])
def debug_create(request):
if not settings.DEBUG:
raise Http404()
form = DebugCreateForm(request.POST or None)
created_link = None
if request.method == "POST" and form.is_valid():
try:
created_link, minted = create_link(
target_url=form.cleaned_data["target_url"],
title=form.cleaned_data.get("title") or "",
external_ref=form.cleaned_data.get("external_ref") or "",
expires_at=None,
token_name="debug",
)
except CodeCollisionError:
form.add_error(None, "Could not allocate a unique code.")
else:
if minted:
messages.success(request, "Short link created.")
else:
messages.info(request, "Existing active link returned (same URL + ref).")
return render(
request,
"links/debug_create.html",
{"form": form, "created_link": created_link},
)
site/links/forms.py
+21
View File
@@ -0,0 +1,21 @@
from django import forms
from links.services import ValidationError, validate_target_url
class DebugCreateForm(forms.Form):
target_url = forms.URLField(
label="Target URL",
widget=forms.URLInput(
attrs={"placeholder": "https://mkdrealtor.com/listings/oak-st", "autofocus": True}
),
)
title = forms.CharField(label="Title", required=False, max_length=200)
external_ref = forms.CharField(label="External ref", required=False, max_length=64)
def clean_target_url(self) -> str:
raw = self.cleaned_data["target_url"]
try:
return validate_target_url(raw)
except ValidationError as exc:
raise forms.ValidationError(str(exc)) from exc
site/links/middleware.py
+68
View File
@@ -0,0 +1,68 @@
"""Keep the short domain and Django admin off the public API hostname."""
from __future__ import annotations
from django.conf import settings
from django.http import Http404, HttpRequest
def _normalize_host(host: str) -> str:
return host.split(":")[0].lower().rstrip(".")
def _host_in(host: str, configured: list[str]) -> bool:
needle = _normalize_host(host)
raw = host.lower()
for entry in configured:
if not entry:
continue
if raw == entry.lower() or needle == _normalize_host(entry):
return True
return False
def is_api_host(host: str) -> bool:
return _host_in(host, list(getattr(settings, "SHORT_API_HOSTS", []) or []))
def is_public_host(host: str) -> bool:
return _host_in(host, list(getattr(settings, "SHORT_PUBLIC_HOSTS", []) or []))
def is_admin_host(host: str) -> bool:
return _host_in(host, list(getattr(settings, "SHORT_ADMIN_HOSTS", []) or []))
class HostSplitMiddleware:
"""Short host = redirects only. API host = /api/ (Bearer). Admin = local only.
A public DNS name may be listed in SHORT_API_HOSTS. Auth, not the network,
keeps /api/ closed: missing/wrong Bearer is 401; empty token list is 503.
"""
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request: HttpRequest):
path = request.path
if path in {"/healthz", "/healthz/"}:
return self.get_response(request)
host = request.get_host()
if path.startswith("/debug"):
if not settings.DEBUG or is_public_host(host):
raise Http404()
return self.get_response(request)
if path.startswith("/admin"):
if not is_admin_host(host):
raise Http404()
return self.get_response(request)
if path.startswith("/api/"):
# Short redirect hostname never serves the API, even if mis-listed.
if is_public_host(host) or not is_api_host(host):
raise Http404()
return self.get_response(request)
site/links/migrations/0001_initial.py
+49
View File
@@ -0,0 +1,49 @@
# Generated by Django 6.1 on 2026-08-30 10:09
import django.db.models.deletion
import uuid
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='ShortLink',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('code', models.CharField(db_index=True, max_length=8, unique=True)),
('target_url', models.URLField(max_length=2048)),
('title', models.CharField(blank=True, max_length=200)),
('created_by_token', models.CharField(max_length=64)),
('external_ref', models.CharField(blank=True, db_index=True, max_length=64)),
('is_active', models.BooleanField(default=True)),
('expires_at', models.DateTimeField(blank=True, null=True)),
('click_count', models.PositiveIntegerField(default=0)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
],
options={
'ordering': ['-created_at'],
},
),
migrations.CreateModel(
name='Click',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('occurred_at', models.DateTimeField(auto_now_add=True)),
('ip_hash', models.CharField(blank=True, max_length=64)),
('user_agent', models.CharField(blank=True, max_length=512)),
('referrer', models.CharField(blank=True, max_length=1024)),
('link', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='clicks', to='links.shortlink')),
],
options={
'ordering': ['-occurred_at'],
},
),
]
site/links/migrations/__init__.py
View File
site/links/models.py
+53
View File
@@ -0,0 +1,53 @@
import uuid
from django.db import models
from django.utils import timezone
class ShortLink(models.Model):
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
code = models.CharField(max_length=8, unique=True, db_index=True)
target_url = models.URLField(max_length=2048)
title = models.CharField(max_length=200, blank=True)
created_by_token = models.CharField(max_length=64)
external_ref = models.CharField(max_length=64, blank=True, db_index=True)
is_active = models.BooleanField(default=True)
expires_at = models.DateTimeField(null=True, blank=True)
click_count = models.PositiveIntegerField(default=0)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
ordering = ["-created_at"]
def __str__(self) -> str:
return self.code
def is_available(self) -> bool:
if not self.is_active:
return False
if self.expires_at is not None and timezone.now() >= self.expires_at:
return False
return True
@property
def public_short_url(self) -> str:
from django.conf import settings
origin = (getattr(settings, "PUBLIC_SHORT_URL", "") or "").rstrip("/")
return f"{origin}/{self.code}"
class Click(models.Model):
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
link = models.ForeignKey(ShortLink, on_delete=models.CASCADE, related_name="clicks")
occurred_at = models.DateTimeField(auto_now_add=True)
ip_hash = models.CharField(max_length=64, blank=True)
user_agent = models.CharField(max_length=512, blank=True)
referrer = models.CharField(max_length=1024, blank=True)
class Meta:
ordering = ["-occurred_at"]
def __str__(self) -> str:
return f"{self.link.code} @ {self.occurred_at}"
site/links/services.py
+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)
site/links/templates/links/debug_create.html
+35
View File
@@ -0,0 +1,35 @@
{% extends "core/base.html" %}
{% block title %}Create short link · debug{% endblock %}
{% block content %}
<p class="eyebrow">Debug only</p>
<h1>Create a short link</h1>
<p class="lede">Uses the same allowlist as the API. Not served when <code>DEBUG</code> is false.</p>
{% if messages %}
<ul class="messages">
{% for message in messages %}<li>{{ message }}</li>{% endfor %}
</ul>
{% endif %}
{% if created_link %}
<div class="result">
<p>Short URL</p>
<p><code>{{ created_link.public_short_url }}</code></p>
<p>Target: {{ created_link.target_url }}</p>
<a class="btn" href="{{ created_link.public_short_url }}">Open</a>
</div>
{% endif %}
<form method="post" action="{% url 'debug-create' %}">
{% csrf_token %}
{{ form.non_field_errors }}
{% for field in form %}
<label>
{{ field.label }}
{{ field }}
{% if field.errors %}<span class="error">{{ field.errors|striptags }}</span>{% endif %}
</label>
{% endfor %}
<button type="submit">Mint</button>
</form>
{% endblock %}
site/links/tests.py
+500
View File
@@ -0,0 +1,500 @@
"""Tests required by Implementation.md §12."""
import json
from datetime import timedelta
from django.contrib.auth import get_user_model
from django.test import TestCase, override_settings
from django.utils import timezone
from django.conf import settings
from links.models import Click, ShortLink
User = get_user_model()
AUTH = "Bearer monica:dev-only-token"
TOKENS = [("monica", "dev-only-token")]
SETTINGS = dict(
SHORTENER_API_TOKENS=TOKENS,
PUBLIC_SHORT_URL="https://go.mkdrealtor.com",
SHORT_PUBLIC_HOSTS=["go.mkdrealtor.com"],
SHORT_API_HOSTS=["testserver", "localhost", "127.0.0.1", "shortener.example.com"],
SHORT_ADMIN_HOSTS=["localhost", "127.0.0.1"],
SHORT_ALLOWED_HOSTS=["mkdrealtor.com"],
CLICK_IP_PEPPER="test-pepper-not-the-secret-key",
ALLOWED_HOSTS=[
"testserver",
"localhost",
"127.0.0.1",
"go.mkdrealtor.com",
"shortener.example.com",
],
SHORT_CODE_LENGTH=6,
)
def _json(response):
return json.loads(response.content.decode())
@override_settings(**SETTINGS)
class AuthTests(TestCase):
def test_missing_bearer_401(self):
response = self.client.post(
"/api/links/",
data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
content_type="application/json",
)
self.assertEqual(response.status_code, 401)
self.assertEqual(response["WWW-Authenticate"], "Bearer")
def test_wrong_token_401(self):
response = self.client.post(
"/api/links/",
data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
content_type="application/json",
HTTP_AUTHORIZATION="Bearer monica:wrong-secret",
)
self.assertEqual(response.status_code, 401)
self.assertEqual(response["WWW-Authenticate"], "Bearer")
self.assertEqual(_json(response)["detail"], "Unauthorized")
def test_matching_named_token_201(self):
response = self.client.post(
"/api/links/",
data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
content_type="application/json",
HTTP_AUTHORIZATION=AUTH,
)
self.assertEqual(response.status_code, 201)
body = _json(response)
self.assertEqual(body["target_url"], "https://mkdrealtor.com/x")
self.assertTrue(body["is_active"])
self.assertEqual(body["click_count"], 0)
@override_settings(SHORTENER_API_TOKENS=[])
def test_empty_tokens_503(self):
response = self.client.post(
"/api/links/",
data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
content_type="application/json",
HTTP_AUTHORIZATION=AUTH,
)
self.assertEqual(response.status_code, 503)
@override_settings(**SETTINGS)
class HostSplitTests(TestCase):
def test_public_host_api_404_even_with_bearer(self):
response = self.client.post(
"/api/links/",
data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
content_type="application/json",
HTTP_AUTHORIZATION=AUTH,
HTTP_HOST="go.mkdrealtor.com",
)
self.assertEqual(response.status_code, 404)
@override_settings(
SHORT_API_HOSTS=[
"testserver",
"localhost",
"127.0.0.1",
"shortener.example.com",
"go.mkdrealtor.com",
]
)
def test_short_host_never_serves_api_even_if_also_listed_as_api(self):
response = self.client.post(
"/api/links/",
data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
content_type="application/json",
HTTP_AUTHORIZATION=AUTH,
HTTP_HOST="go.mkdrealtor.com",
)
self.assertEqual(response.status_code, 404)
def test_public_host_admin_404(self):
response = self.client.get("/admin/", HTTP_HOST="go.mkdrealtor.com")
self.assertEqual(response.status_code, 404)
def test_public_api_host_without_bearer_401(self):
response = self.client.post(
"/api/links/",
data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
content_type="application/json",
HTTP_HOST="shortener.example.com",
)
self.assertEqual(response.status_code, 401)
self.assertEqual(response["WWW-Authenticate"], "Bearer")
def test_public_api_host_wrong_token_401(self):
response = self.client.post(
"/api/links/",
data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
content_type="application/json",
HTTP_AUTHORIZATION="Bearer monica:wrong-secret",
HTTP_HOST="shortener.example.com",
)
self.assertEqual(response.status_code, 401)
def test_public_api_host_valid_bearer_201(self):
response = self.client.post(
"/api/links/",
data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
content_type="application/json",
HTTP_AUTHORIZATION=AUTH,
HTTP_HOST="shortener.example.com",
)
self.assertEqual(response.status_code, 201)
def test_public_api_host_admin_404(self):
response = self.client.get("/admin/", HTTP_HOST="shortener.example.com")
self.assertEqual(response.status_code, 404)
def test_healthz_on_public_and_api(self):
for host in ("go.mkdrealtor.com", "testserver", "shortener.example.com"):
response = self.client.get("/healthz/", HTTP_HOST=host)
self.assertEqual(response.status_code, 200)
self.assertEqual(_json(response), {"status": "ok"})
@override_settings(**SETTINGS)
class AllowlistTests(TestCase):
def _post(self, target_url):
return self.client.post(
"/api/links/",
data=json.dumps({"target_url": target_url}),
content_type="application/json",
HTTP_AUTHORIZATION=AUTH,
)
def test_https_allowlisted_ok(self):
self.assertEqual(self._post("https://mkdrealtor.com/x").status_code, 201)
self.assertEqual(self._post("https://www.mkdrealtor.com/x").status_code, 201)
@override_settings(SHORT_ALLOWED_HOSTS=["mkdrealtor.com", "*.aimloperations.com"])
def test_glob_suffix_and_apex(self):
self.assertEqual(
self._post("https://aimloperations.com/web_design").status_code, 201
)
self.assertEqual(
self._post("https://www.aimloperations.com/web_design").status_code, 201
)
def test_http_rejected(self):
self.assertEqual(self._post("http://mkdrealtor.com/x").status_code, 400)
def test_evil_host_rejected(self):
self.assertEqual(self._post("https://evil.com").status_code, 400)
def test_javascript_rejected(self):
self.assertEqual(self._post("javascript:alert(1)").status_code, 400)
def test_protocol_relative_rejected(self):
self.assertEqual(self._post("//evil.com").status_code, 400)
def test_credentials_rejected(self):
self.assertEqual(
self._post("https://user:pass@mkdrealtor.com/x").status_code, 400
)
def test_suffix_does_not_match_cousin_domain(self):
self.assertEqual(self._post("https://notmkdrealtor.com/x").status_code, 400)
@override_settings(**SETTINGS)
class RedirectTests(TestCase):
def setUp(self):
self.link = ShortLink.objects.create(
code="a3k9xm",
target_url="https://mkdrealtor.com/listings/oak-st",
created_by_token="monica",
)
def test_active_code_302_to_target(self):
response = self.client.get(
"/a3k9xm", HTTP_HOST="go.mkdrealtor.com", follow=False
)
self.assertEqual(response.status_code, 302)
self.assertNotEqual(response.status_code, 301)
self.assertEqual(
response["Location"], "https://mkdrealtor.com/listings/oak-st"
)
def test_head_also_302_without_click(self):
response = self.client.head(
"/a3k9xm", HTTP_HOST="go.mkdrealtor.com", follow=False
)
self.assertEqual(response.status_code, 302)
self.assertEqual(
response["Location"], "https://mkdrealtor.com/listings/oak-st"
)
self.link.refresh_from_db()
self.assertEqual(self.link.click_count, 0)
self.assertEqual(Click.objects.filter(link=self.link).count(), 0)
def test_inactive_404(self):
self.link.is_active = False
self.link.save()
response = self.client.get("/a3k9xm", HTTP_HOST="go.mkdrealtor.com")
self.assertEqual(response.status_code, 404)
def test_expired_404(self):
self.link.expires_at = timezone.now() - timedelta(minutes=1)
self.link.save()
response = self.client.get("/a3k9xm", HTTP_HOST="go.mkdrealtor.com")
self.assertEqual(response.status_code, 404)
def test_unknown_404(self):
response = self.client.get("/zzzzzz", HTTP_HOST="go.mkdrealtor.com")
self.assertEqual(response.status_code, 404)
def test_invalid_code_shape_404(self):
response = self.client.get("/AB", HTTP_HOST="go.mkdrealtor.com")
self.assertEqual(response.status_code, 404)
def test_public_root_is_landing(self):
response = self.client.get("/", HTTP_HOST="go.mkdrealtor.com")
self.assertEqual(response.status_code, 200)
self.assertContains(response, "URL Shortening Service")
self.assertContains(response, "aimloperations.com")
self.assertNotContains(response, "Create a short link")
def test_click_row_and_count(self):
self.client.get(
"/a3k9xm",
HTTP_HOST="go.mkdrealtor.com",
HTTP_USER_AGENT="sms-client",
)
self.link.refresh_from_db()
self.assertEqual(self.link.click_count, 1)
self.assertEqual(Click.objects.filter(link=self.link).count(), 1)
click = Click.objects.get(link=self.link)
self.assertEqual(click.user_agent, "sms-client")
self.assertTrue(click.ip_hash)
self.assertNotIn("127.0.0.1", click.ip_hash)
@override_settings(**SETTINGS)
class CreateTests(TestCase):
def _post(self, payload, **headers):
return self.client.post(
"/api/links/",
data=json.dumps(payload),
content_type="application/json",
HTTP_AUTHORIZATION=AUTH,
**headers,
)
def test_short_url_uses_public_origin_not_api_host(self):
response = self._post(
{"target_url": "https://mkdrealtor.com/x"},
HTTP_HOST="testserver",
)
self.assertEqual(response.status_code, 201)
body = _json(response)
self.assertTrue(body["short_url"].startswith("https://go.mkdrealtor.com/"))
self.assertNotIn("testserver", body["short_url"])
def test_idempotent_same_target_and_external_ref(self):
payload = {
"target_url": "https://mkdrealtor.com/listings/oak-st?utm_source=monica",
"title": "Oak St",
"external_ref": "campaign-1",
}
first = self._post(payload)
second = self._post(payload)
self.assertEqual(first.status_code, 201)
self.assertEqual(second.status_code, 200)
self.assertEqual(_json(first)["code"], _json(second)["code"])
self.assertEqual(ShortLink.objects.count(), 1)
def test_empty_external_ref_always_mints(self):
payload = {"target_url": "https://mkdrealtor.com/x"}
first = self._post(payload)
second = self._post(payload)
self.assertEqual(first.status_code, 201)
self.assertEqual(second.status_code, 201)
self.assertNotEqual(_json(first)["code"], _json(second)["code"])
def test_code_uses_unambiguous_alphabet(self):
alphabet = set(settings.CODE_ALPHABET)
forbidden = set("01iloILO")
for _ in range(20):
response = self._post({"target_url": "https://mkdrealtor.com/x"})
code = _json(response)["code"]
self.assertTrue(set(code) <= alphabet)
self.assertFalse(set(code) & forbidden)
self.assertEqual(len(code), 6)
def test_stores_token_name_not_secret(self):
self._post({"target_url": "https://mkdrealtor.com/x"})
link = ShortLink.objects.get()
self.assertEqual(link.created_by_token, "monica")
self.assertNotIn("dev-only-token", link.created_by_token)
def test_disable_is_idempotent(self):
code = _json(self._post({"target_url": "https://mkdrealtor.com/x"}))["code"]
first = self.client.post(
f"/api/links/{code}/disable/",
HTTP_AUTHORIZATION=AUTH,
)
second = self.client.post(
f"/api/links/{code}/disable/",
HTTP_AUTHORIZATION=AUTH,
)
self.assertEqual(first.status_code, 200)
self.assertEqual(second.status_code, 200)
self.assertFalse(_json(second)["is_active"])
self.client.get(f"/{code}", HTTP_HOST="go.mkdrealtor.com")
# disabled → no redirect
response = self.client.get(f"/{code}", HTTP_HOST="go.mkdrealtor.com")
self.assertEqual(response.status_code, 404)
@override_settings(**SETTINGS)
class UseCaseTests(TestCase):
"""monica_site mints via Bearer; a phone hits the public short URL and 302s."""
def test_caller_creates_then_public_redirects(self):
create = self.client.post(
"/api/links/",
data=json.dumps(
{
"target_url": (
"https://mkdrealtor.com/listings/oak-st"
"?utm_source=monica&utm_medium=sms"
),
"title": "Oak St listing",
"external_ref": "campaign-uuid-optional",
}
),
content_type="application/json",
HTTP_AUTHORIZATION="Bearer monica:dev-only-token",
HTTP_HOST="testserver",
)
self.assertEqual(create.status_code, 201)
body = _json(create)
self.assertEqual(
body["short_url"], f"https://go.mkdrealtor.com/{body['code']}"
)
follow = self.client.get(
f"/{body['code']}",
HTTP_HOST="go.mkdrealtor.com",
follow=False,
)
self.assertEqual(follow.status_code, 302)
self.assertEqual(
follow["Location"],
"https://mkdrealtor.com/listings/oak-st?utm_source=monica&utm_medium=sms",
)
@override_settings(**SETTINGS)
class DebugCreateTests(TestCase):
def test_hidden_when_not_debug(self):
response = self.client.get("/debug/")
self.assertEqual(response.status_code, 404)
@override_settings(DEBUG=True)
def test_form_when_debug(self):
response = self.client.get("/debug/")
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Create a short link")
@override_settings(DEBUG=True)
def test_hidden_on_public_short_host_even_in_debug(self):
response = self.client.get("/debug/", HTTP_HOST="go.mkdrealtor.com")
self.assertEqual(response.status_code, 404)
@override_settings(DEBUG=True)
def test_post_mints_link(self):
response = self.client.post(
"/debug/",
{
"target_url": "https://mkdrealtor.com/listings/oak-st",
"title": "Oak St",
},
)
self.assertEqual(response.status_code, 200)
self.assertEqual(ShortLink.objects.count(), 1)
link = ShortLink.objects.get()
self.assertEqual(link.created_by_token, "debug")
self.assertContains(response, link.public_short_url)
ADMIN_SETTINGS = {**SETTINGS, "SHORT_ADMIN_HOSTS": ["testserver", "localhost"]}
@override_settings(**ADMIN_SETTINGS)
class AdminTests(TestCase):
def setUp(self):
self.user = User.objects.create_superuser("admin", "admin@example.com", "pass")
self.client.force_login(self.user)
self.link = ShortLink.objects.create(
code="a3k9xm",
target_url="https://mkdrealtor.com/listings/oak-st",
title="Oak St",
created_by_token="monica",
)
def test_changelist(self):
response = self.client.get("/admin/links/shortlink/")
self.assertEqual(response.status_code, 200)
self.assertContains(response, "a3k9xm")
self.assertContains(response, "Oak St")
def test_add_page(self):
response = self.client.get("/admin/links/shortlink/add/")
self.assertEqual(response.status_code, 200)
self.assertContains(response, "Target url")
def test_add_mints_code(self):
response = self.client.post(
"/admin/links/shortlink/add/",
{
"target_url": "https://mkdrealtor.com/new",
"title": "From admin",
"is_active": "on",
"external_ref": "",
"clicks-TOTAL_FORMS": "0",
"clicks-INITIAL_FORMS": "0",
"clicks-MIN_NUM_FORMS": "0",
"clicks-MAX_NUM_FORMS": "0",
"_save": "Save",
},
)
self.assertEqual(response.status_code, 302)
created = ShortLink.objects.exclude(code="a3k9xm").get()
self.assertEqual(created.target_url, "https://mkdrealtor.com/new")
self.assertEqual(created.created_by_token, "admin")
self.assertEqual(len(created.code), 6)
def test_disable_action(self):
response = self.client.post(
"/admin/links/shortlink/",
{
"action": "disable_links",
"_selected_action": [str(self.link.pk)],
},
)
self.assertEqual(response.status_code, 302)
self.link.refresh_from_db()
self.assertFalse(self.link.is_active)
def test_click_changelist(self):
Click.objects.create(link=self.link, user_agent="sms-client")
response = self.client.get("/admin/links/click/")
self.assertEqual(response.status_code, 200)
self.assertContains(response, "a3k9xm")
def test_admin_404_on_public_hosts(self):
self.client.logout()
response = self.client.get("/admin/", HTTP_HOST="go.mkdrealtor.com")
self.assertEqual(response.status_code, 404)
response = self.client.get("/admin/", HTTP_HOST="shortener.example.com")
self.assertEqual(response.status_code, 404)
site/links/urls.py
+9
View File
@@ -0,0 +1,9 @@
from django.urls import path
from links import api
urlpatterns = [
path("", api.links_collection, name="links-collection"),
path("<str:code>/", api.link_detail, name="links-detail"),
path("<str:code>/disable/", api.link_disable, name="links-disable"),
]
site/links/views.py
+31
View File
@@ -0,0 +1,31 @@
"""Public redirect — no auth. SMS recipients tap GET /<code>."""
from __future__ import annotations
import re
from django.http import Http404, HttpResponseRedirect
from django.views.decorators.http import require_http_methods
from links.models import ShortLink
from links.services import record_click_best_effort
CODE_RE = re.compile(r"^[a-z0-9]{4,8}$")
@require_http_methods(["GET", "HEAD"])
def redirect_view(request, code: str):
if not CODE_RE.fullmatch(code):
raise Http404()
try:
link = ShortLink.objects.get(code=code)
except ShortLink.DoesNotExist as exc:
raise Http404() from exc
if not link.is_available():
raise Http404()
if request.method == "GET":
record_click_best_effort(request, link)
return HttpResponseRedirect(link.target_url)
site/manage.py
+23
View File
@@ -0,0 +1,23 @@
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys
def main() -> None:
"""Run administrative tasks."""
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "shortener.settings")
try:
from django.core.management import execute_from_command_line
except ImportError as exc:
raise ImportError(
"Couldn't import Django. Are you sure it's installed and "
"available on your PYTHONPATH environment variable? Did you "
"forget to activate a virtual environment?"
) from exc
execute_from_command_line(sys.argv)
if __name__ == "__main__":
main()
site/shortener/__init__.py
+1
View File
@@ -0,0 +1 @@
site/shortener/asgi.py
+9
View File
@@ -0,0 +1,9 @@
"""ASGI config for shortener."""
import os
from django.core.asgi import get_asgi_application
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "shortener.settings")
application = get_asgi_application()
site/shortener/settings/__init__.py
+12
View File
@@ -0,0 +1,12 @@
"""Load environment-specific Django settings based on DJANGO_ENV."""
import os
_environment = os.environ.get("DJANGO_ENV", "dev").lower()
if _environment == "prod":
from .prod import * # noqa: F403
elif _environment == "beta":
from .beta import * # noqa: F403
else:
from .dev import * # noqa: F403
site/shortener/settings/base.py
+198
View File
@@ -0,0 +1,198 @@
"""Shared Django settings for all environments."""
import json
import os
from pathlib import Path
from urllib.parse import urlparse
BASE_DIR = Path(__file__).resolve().parent.parent.parent
def env(key: str, default: str | None = None) -> str | None:
return os.environ.get(key, default)
def env_bool(key: str, default: bool = False) -> bool:
value = os.environ.get(key)
if value is None:
return default
return value.lower() in {"1", "true", "yes", "on"}
def env_list(key: str, default: str = "") -> list[str]:
value = os.environ.get(key, default)
if not value:
return []
value = value.strip()
if value.startswith("["):
try:
parsed = json.loads(value)
except ValueError:
parsed = None
if isinstance(parsed, list):
return [str(item).strip() for item in parsed if str(item).strip()]
return [item.strip() for item in value.split(",") if item.strip()]
def env_int(key: str, default: int) -> int:
value = env(key)
if value is None or value == "":
return default
return int(value)
def parse_api_tokens(raw: str) -> list[tuple[str, str]]:
"""Parse ``name:secret,name:secret`` into ``[(name, secret), ...]``."""
tokens: list[tuple[str, str]] = []
if not raw:
return tokens
for part in raw.split(","):
part = part.strip()
if not part or ":" not in part:
continue
name, secret = part.split(":", 1)
name, secret = name.strip(), secret.strip()
if name and secret:
tokens.append((name, secret))
return tokens
def database_config() -> dict:
database_url = env("DATABASE_URL")
if database_url:
parsed = urlparse(database_url)
return {
"default": {
"ENGINE": "django.db.backends.postgresql",
"NAME": parsed.path.lstrip("/"),
"USER": parsed.username or "",
"PASSWORD": parsed.password or "",
"HOST": parsed.hostname or "",
"PORT": str(parsed.port or 5432),
}
}
return {
"default": {
"ENGINE": "django.db.backends.sqlite3",
"NAME": BASE_DIR / "db.sqlite3",
}
}
SECRET_KEY = env(
"DJANGO_SECRET_KEY",
"django-insecure-dev-only-change-me-before-production",
)
DEBUG = env_bool("DJANGO_DEBUG", False)
allowed_hosts = env_list(
"DJANGO_ALLOWED_HOSTS",
"localhost,127.0.0.1,0.0.0.0,testserver,web,url-shortener,go.mkdrealtor.com",
)
ALLOWED_HOSTS = allowed_hosts if allowed_hosts else ["*"]
INSTALLED_APPS = [
"core.apps.CoreConfig",
"links.apps.LinksConfig",
"django.contrib.admin",
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.sessions",
"django.contrib.messages",
"whitenoise.runserver_nostatic",
"django.contrib.staticfiles",
]
MIDDLEWARE = [
"django.middleware.security.SecurityMiddleware",
"whitenoise.middleware.WhiteNoiseMiddleware",
"links.middleware.HostSplitMiddleware",
"django.contrib.sessions.middleware.SessionMiddleware",
"django.middleware.common.CommonMiddleware",
"django.middleware.csrf.CsrfViewMiddleware",
"django.contrib.auth.middleware.AuthenticationMiddleware",
"django.contrib.messages.middleware.MessageMiddleware",
"django.middleware.clickjacking.XFrameOptionsMiddleware",
]
ROOT_URLCONF = "shortener.urls"
TEMPLATES = [
{
"BACKEND": "django.template.backends.django.DjangoTemplates",
"DIRS": [],
"APP_DIRS": True,
"OPTIONS": {
"context_processors": [
"django.template.context_processors.debug",
"django.template.context_processors.request",
"django.contrib.auth.context_processors.auth",
"django.contrib.messages.context_processors.messages",
"core.context_processors.branding",
],
},
},
]
WSGI_APPLICATION = "shortener.wsgi.application"
DATABASES = database_config()
AUTH_PASSWORD_VALIDATORS = [
{
"NAME": "django.contrib.auth.password_validation.UserAttributeSimilarityValidator",
},
{
"NAME": "django.contrib.auth.password_validation.MinimumLengthValidator",
},
{
"NAME": "django.contrib.auth.password_validation.CommonPasswordValidator",
},
{
"NAME": "django.contrib.auth.password_validation.NumericPasswordValidator",
},
]
LANGUAGE_CODE = "en-us"
TIME_ZONE = "America/Chicago"
USE_I18N = True
USE_TZ = True
STATIC_URL = "static/"
STATIC_ROOT = BASE_DIR / "staticfiles"
STORAGES = {
"default": {
"BACKEND": "django.core.files.storage.memory.InMemoryStorage",
},
"staticfiles": {
"BACKEND": "whitenoise.storage.CompressedManifestStaticFilesStorage",
},
}
DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
# --- Shortener ---
SHORT_DOMAIN = env("SHORT_DOMAIN", "localhost:8000") or "localhost:8000"
PUBLIC_SHORT_URL = (env("PUBLIC_SHORT_URL", "https://go.mkdrealtor.com") or "").rstrip(
"/"
)
SHORT_PUBLIC_HOSTS = env_list("SHORT_PUBLIC_HOSTS", SHORT_DOMAIN.split(":")[0])
SHORT_API_HOSTS = env_list(
"SHORT_API_HOSTS",
"localhost,127.0.0.1,0.0.0.0,testserver,web,url-shortener",
)
# Django admin — local/dev only. Never put the public API hostname here.
SHORT_ADMIN_HOSTS = env_list("SHORT_ADMIN_HOSTS", "localhost,127.0.0.1")
SHORTENER_API_TOKENS = parse_api_tokens(env("SHORTENER_API_TOKENS", "") or "")
SHORT_ALLOWED_HOSTS = env_list(
"SHORT_ALLOWED_HOSTS", "mkdrealtor.com,aimloperations.com"
)
SHORT_CODE_LENGTH = env_int("SHORT_CODE_LENGTH", 6)
CLICK_IP_PEPPER = env("CLICK_IP_PEPPER", "") or ""
CODE_ALPHABET = "23456789abcdefghjkmnpqrstuvwxyz"
SITE_NAME = env("SITE_NAME", "URL Shortening Service") or "URL Shortening Service"
CREDIT_NAME = env("CREDIT_NAME", "AI ML Operations") or "AI ML Operations"
CREDIT_URL = env("CREDIT_URL", "https://aimloperations.com") or "https://aimloperations.com"
site/shortener/settings/beta.py
+18
View File
@@ -0,0 +1,18 @@
"""Beta/staging settings."""
from .base import * # noqa: F403
from .logging import build_logging_config, logging_level_for_env
DEBUG = env_bool("DJANGO_DEBUG", False) # noqa: F405
if DEBUG:
import warnings
warnings.warn("DEBUG is enabled in beta environment.", stacklevel=1)
SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")
USE_X_FORWARDED_HOST = True
SESSION_COOKIE_SECURE = not DEBUG
CSRF_COOKIE_SECURE = not DEBUG
LOGGING = build_logging_config(logging_level_for_env("beta"), "beta")
site/shortener/settings/dev.py
+17
View File
@@ -0,0 +1,17 @@
"""Development settings."""
from .base import * # noqa: F403
from .logging import build_logging_config, logging_level_for_env
DEBUG = True
STORAGES = {
"default": {
"BACKEND": "django.core.files.storage.memory.InMemoryStorage",
},
"staticfiles": {
"BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage",
},
}
LOGGING = build_logging_config(logging_level_for_env("dev"), "dev")
site/shortener/settings/logging.py
+82
View File
@@ -0,0 +1,82 @@
"""Environment-specific logging configuration."""
import os
def build_logging_config(level: str, environment: str) -> dict:
"""Return a Django LOGGING dict for the given level and environment name."""
return {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"verbose": {
"format": (
f"{{levelname}} {{asctime}} {{name}} {{filename}}:{{lineno}} "
f"{{process:d}} {{thread:d}} [env={environment}] {{message}}"
),
"style": "{",
},
"simple": {
"format": (
f"{{levelname}} [env={environment}] "
f"{{filename}}:{{lineno}} {{message}}"
),
"style": "{",
},
},
"filters": {
"strip_authorization": {
"()": "shortener.settings.logging.StripAuthorizationFilter",
},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "verbose" if environment == "dev" else "simple",
"filters": ["strip_authorization"],
},
},
"root": {
"handlers": ["console"],
"level": level,
},
"loggers": {
"django": {
"handlers": ["console"],
"level": level,
"propagate": False,
},
"django.request": {
"handlers": ["console"],
"level": "ERROR" if environment == "prod" else level,
"propagate": False,
},
"django.server": {
"handlers": ["console"],
"level": level,
"propagate": False,
},
},
}
class StripAuthorizationFilter:
"""Drop log records that appear to contain an Authorization header."""
def filter(self, record) -> bool:
message = record.getMessage()
if "authorization" in message.lower() and "bearer" in message.lower():
return False
return True
def logging_level_for_env(environment: str) -> str:
override = os.environ.get("DJANGO_LOG_LEVEL")
if override:
return override.upper()
if environment == "dev":
return "DEBUG"
if environment == "beta":
return "INFO"
return "WARNING"
site/shortener/settings/prod.py
+16
View File
@@ -0,0 +1,16 @@
"""Production settings."""
from .base import * # noqa: F403
from .logging import build_logging_config, logging_level_for_env
DEBUG = False
if not env("DJANGO_SECRET_KEY"): # noqa: F405
raise ValueError("DJANGO_SECRET_KEY must be set in production.")
SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")
USE_X_FORWARDED_HOST = True
SESSION_COOKIE_SECURE = True
CSRF_COOKIE_SECURE = True
LOGGING = build_logging_config(logging_level_for_env("prod"), "prod")
site/shortener/urls.py
+17
View File
@@ -0,0 +1,17 @@
"""URL configuration for shortener."""
from django.contrib import admin
from django.urls import include, path
from core.views import healthz, landing
from links.debug_views import debug_create
from links.views import redirect_view
urlpatterns = [
path("", landing, name="landing"),
path("healthz/", healthz, name="healthz"),
path("debug/", debug_create, name="debug-create"),
path("admin/", admin.site.urls),
path("api/links/", include("links.urls")),
path("<str:code>", redirect_view, name="redirect"),
]
site/shortener/wsgi.py
+9
View File
@@ -0,0 +1,9 @@
"""WSGI config for shortener."""
import os
from django.core.wsgi import get_wsgi_application
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "shortener.settings")
application = get_wsgi_application()