From 14d6e8c50e6757b2ad90d3c17d53f8fafc87a35a Mon Sep 17 00:00:00 2001 From: Jason Stedwell Date: Fri, 3 Jul 2026 01:39:50 -0500 Subject: [PATCH] feat: per-link redirect type, password-protected links, branded visitor pages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Redirects default to 302 so browsers stop permanently caching the hop — click analytics stay accurate and destination edits take effect for returning visitors. 301 is a per-link opt-in (create/edit/API). - Links can carry a password: visitors get a branded unlock page at /:code and POST to /:code/unlock. Unlocks always redirect with 302. - Missing, expired, and quota-limited links now show branded standalone visitor pages (404/410/429) instead of bouncing to the marketing site with ?error= query params. - Clicks are recorded through the async pipeline with bot/crawler tagging (UA heuristics), GeoLite2 country lookup (ip-api fallback gated behind GEO_API_FALLBACK), and the IP_ANONYMIZE policy applied at write time. - Deletes stamp deleted_at for later hard-purge. - Dashboard: redirect-type selector and password set/replace/remove in the edit form, lock badge on protected links. Co-Authored-By: Claude Fable 5 --- app.py | 276 ++++++++++++++++++++++++++++++++++++++++++----------- index.html | 35 ++++++- 2 files changed, 252 insertions(+), 59 deletions(-) diff --git a/app.py b/app.py index b8093ff..9d266bc 100644 --- a/app.py +++ b/app.py @@ -637,6 +637,21 @@ def parse_referrer(ref): if 'instagram' in r: return 'Instagram' return 'Other' +_BOT_UA_MARKERS = ( + 'bot', 'crawler', 'spider', 'slurp', 'curl', 'wget', 'python-requests', + 'python-urllib', 'httpclient', 'go-http-client', 'okhttp', 'scrapy', + 'headless', 'phantomjs', 'facebookexternalhit', 'whatsapp', 'telegrambot', + 'discordbot', 'slackbot', 'linkedinbot', 'twitterbot', 'pinterest', + 'preview', 'monitor', 'pingdom', 'uptime', 'postmanruntime', 'java/', +) + +def is_bot_ua(ua): + """Heuristic bot/crawler detection from the User-Agent string.""" + if not ua: + return True + u = ua.lower() + return any(m in u for m in _BOT_UA_MARKERS) + def get_client_ip(): """Return the real client IP, honouring X-Forwarded-For from trusted proxies.""" xff = request.headers.get('X-Forwarded-For', '') @@ -644,42 +659,71 @@ def get_client_ip(): return xff.split(',')[0].strip() return request.remote_addr or '' -def get_country_for_request(): - """Return a 2-letter ISO country code for the current request. +def anonymize_ip(ip): + """Apply the configured IP_ANONYMIZE policy before an IP is stored. + 'truncate' zeroes the host part; 'hash' keeps unique-visitor counting + possible without retaining the address itself.""" + if not ip or IP_ANONYMIZE == 'none': + return ip + if IP_ANONYMIZE == 'hash': + return hashlib.sha256(ip.encode()).hexdigest()[:16] + if IP_ANONYMIZE == 'truncate': + if ':' in ip: + parts = ip.split(':') + return ':'.join(parts[:3]) + '::' + parts = ip.split('.') + if len(parts) == 4: + return '.'.join(parts[:3]) + '.0' + return ip + +# Local GeoLite2 country database (preferred over any network lookup). +# Download GeoLite2-Country.mmdb from MaxMind and mount it at GEOIP_DB_PATH. +_geoip_reader = None +try: + import geoip2.database as _geoip2_db + if os.path.exists(GEOIP_DB_PATH): + _geoip_reader = _geoip2_db.Reader(GEOIP_DB_PATH) +except Exception: + _geoip_reader = None + +def get_country_for_ip(ip): + """Return a 2-letter ISO country code for an IP address. Priority: - 1. CF-IPCountry header (Cloudflare — zero-latency, most reliable) - 2. ip-api.com free JSON API (1-second timeout, fails gracefully) - Returns 'XX' if the IP is private/loopback, 'Unknown' on any failure. + 1. Local MaxMind GeoLite2 database (offline, no rate limits, commercial-safe) + 2. ip-api.com free JSON API — only if GEO_API_FALLBACK=true (their free + tier is rate-limited and licensed for non-commercial use only) + Returns 'XX' for private/loopback IPs, 'Unknown' on any failure. + Called from the background click writer, never from the request path. """ - import urllib.request as _ureq, json as _json - - cf = request.headers.get('CF-IPCountry', '').strip().upper() - if cf and len(cf) == 2 and cf.isalpha() and cf != 'XX': - return cf - - ip = get_client_ip() - if not ip or ip in ('127.0.0.1', '::1'): - return 'XX' - # Skip RFC-1918 / loopback ranges — no point querying for private IPs + if not ip: + return 'Unknown' try: import ipaddress parsed = ipaddress.ip_address(ip) if parsed.is_private or parsed.is_loopback or parsed.is_link_local: return 'XX' except ValueError: - pass - try: - req = _ureq.Request( - f'http://ip-api.com/json/{ip}?fields=countryCode', - headers={'User-Agent': 'QRknit/1.0'} - ) - with _ureq.urlopen(req, timeout=1) as resp: - data = _json.loads(resp.read()) - code = data.get('countryCode', '') - return code if code else 'Unknown' - except Exception: return 'Unknown' + if _geoip_reader: + try: + return _geoip_reader.country(ip).country.iso_code or 'Unknown' + except Exception: + return 'Unknown' + if GEO_API_FALLBACK: + try: + import urllib.request as _ureq + req = _ureq.Request( + f'http://ip-api.com/json/{ip}?fields=countryCode', + headers={'User-Agent': 'QRknit/1.0'} + ) + with _ureq.urlopen(req, timeout=2) as resp: + data = json.loads(resp.read()) + code = data.get('countryCode', '') + return code if code else 'Unknown' + except Exception: + return 'Unknown' + return 'Unknown' def get_link_tags(conn, link_id): rows = conn.execute( @@ -718,6 +762,8 @@ def format_link(row, conn): 'clicks': row['clicks'], 'is_active': row['is_active'], 'is_pinned': row['is_pinned'], + 'redirect_type': row['redirect_type'] or 302, + 'has_password': bool(row['password_hash']), 'short_url': f"{BASE_URL}/{row['code']}", 'qr_url': f"{BASE_URL}/api/qr/{row['code']}", 'tags': get_link_tags(conn, row['id']), @@ -849,12 +895,14 @@ def me(): @app.route('/api/shorten', methods=['POST']) @login_required def shorten(): - data = request.get_json(silent=True) or {} - long_url = (data.get('url') or '').strip() - custom_code = (data.get('custom_code') or '').strip() - title = (data.get('title') or '').strip() - expires_at = data.get('expires_at') - tags = data.get('tags', []) + data = request.get_json(silent=True) or {} + long_url = (data.get('url') or '').strip() + custom_code = (data.get('custom_code') or '').strip() + title = (data.get('title') or '').strip() + expires_at = data.get('expires_at') + tags = data.get('tags', []) + redirect_type = data.get('redirect_type', 302) + password = data.get('password') or '' if not long_url: return jsonify({'error': 'URL is required'}), 400 @@ -862,6 +910,9 @@ def shorten(): return jsonify({'error': 'URL must start with http:// or https://'}), 400 if custom_code and not re.match(r'^[a-zA-Z0-9]{1,20}$', custom_code): return jsonify({'error': 'Custom code must be 1–20 alphanumeric characters'}), 400 + if redirect_type not in (301, 302, '301', '302'): + return jsonify({'error': 'redirect_type must be 301 or 302'}), 400 + redirect_type = int(redirect_type) code = custom_code or generate_code(long_url) @@ -883,10 +934,12 @@ def shorten(): code = generate_code(long_url + str(time.time())) conn.execute( - 'INSERT INTO links (code,long_url,title,created_at,expires_at,user_id) VALUES (?,?,?,?,?,?)', + 'INSERT INTO links (code,long_url,title,created_at,expires_at,user_id,redirect_type,password_hash) ' + 'VALUES (?,?,?,?,?,?,?,?)', (code, long_url, title or None, datetime.now(timezone.utc).replace(tzinfo=None).isoformat(), - expires_at or None, session.get('user_id')) + expires_at or None, session.get('user_id'), redirect_type, + generate_password_hash(password) if password else None) ) link_id = conn.execute('SELECT id FROM links WHERE code=?', (code,)).fetchone()['id'] if tags: @@ -898,6 +951,8 @@ def shorten(): 'long_url': long_url, 'title': title, 'tags': tags, + 'redirect_type': redirect_type, + 'has_password': bool(password), 'qr_url': f"{BASE_URL}/api/qr/{code}", }), 201 @@ -1005,6 +1060,14 @@ def edit_link(code): if 'title' in data: updates['title'] = data['title'].strip() or None if 'expires_at' in data: updates['expires_at'] = data['expires_at'] or None if 'is_pinned' in data: updates['is_pinned'] = 1 if data['is_pinned'] else 0 + if 'redirect_type' in data: + if data['redirect_type'] not in (301, 302, '301', '302'): + return jsonify({'error': 'redirect_type must be 301 or 302'}), 400 + updates['redirect_type'] = int(data['redirect_type']) + if 'password' in data: + # Non-empty string sets/updates the password; empty/null removes it + pw = data['password'] or '' + updates['password_hash'] = generate_password_hash(pw) if pw else None if updates: set_clause = ', '.join(f'{k}=?' for k in updates) @@ -1024,7 +1087,7 @@ def delete_link(code): link = conn.execute('SELECT * FROM links WHERE code=?', (code,)).fetchone() if not link or not _can_access_link(link, conn): return jsonify({'error': 'Not found'}), 404 - conn.execute('UPDATE links SET is_active=0 WHERE code=?', (code,)) + conn.execute('UPDATE links SET is_active=0, deleted_at=? WHERE code=?', (now_iso(), code)) return jsonify({'success': True}) @@ -1380,11 +1443,12 @@ def bulk_links(): with get_db() as conn: if action == 'delete': if is_admin: - conn.execute(f'UPDATE links SET is_active=0 WHERE code IN ({placeholders})', codes) + conn.execute(f'UPDATE links SET is_active=0, deleted_at=? WHERE code IN ({placeholders})', + [now_iso()] + list(codes)) else: conn.execute( - f'UPDATE links SET is_active=0 WHERE code IN ({placeholders}) AND user_id=?', - list(codes) + [user_id] + f'UPDATE links SET is_active=0, deleted_at=? WHERE code IN ({placeholders}) AND user_id=?', + [now_iso()] + list(codes) + [user_id] ) return jsonify({'deleted': len(codes)}) @@ -1810,9 +1874,95 @@ def sitemap_xml(): # ───────────────────────────────────────────── -# Redirect +# Redirect — branded visitor pages, password gate, async click logging # ───────────────────────────────────────────── +def _visitor_page(heading, message, body_html='', status=200): + """Branded standalone page shown to link visitors (expired, missing, + password-protected, …) — they never see the app or marketing site.""" + html = f""" + +{heading} — {APP_NAME} + +
+
▮ {APP_NAME}
+

{heading}

+

{message}

+ {body_html} +
+""" + return Response(html, status=status, mimetype='text/html', + headers={'Cache-Control': 'no-store'}) + + +def _password_form(code, error=''): + err_html = f'
{error}
' if error else '' + return _visitor_page( + 'This link is password protected', + 'Enter the password to continue to the destination.', + f'''
+ + +
{err_html}''', + status=200 + ) + + +def _record_click(link_row): + """Capture request details and enqueue the click for the background writer.""" + ua = request.headers.get('User-Agent', '')[:500] + cf = request.headers.get('CF-IPCountry', '').strip().upper() + item = { + 'link_id': link_row['id'], + 'ts': now_iso(), + 'referrer': request.referrer, + 'ua': ua, + 'ip': get_client_ip()[:45], + 'cf_country': cf if (len(cf) == 2 and cf.isalpha() and cf != 'XX') else '', + 'is_bot': 1 if is_bot_ua(ua) else 0, + } + try: + _click_queue.put_nowait(item) + except queue.Full: + pass + + +def _check_link_gates(conn, link): + """Expiry and quota checks shared by the redirect and unlock routes. + Returns an error Response, or None if the link may be followed.""" + if link['expires_at'] and link['expires_at'] < now_iso(): + return _visitor_page('Link expired', + 'This short link has passed its expiry date and is no longer available.', + status=410) + if link['user_id']: + plan = get_user_plan(conn, link['user_id']) + click_limit = PLAN_LIMITS[plan]['monthly_clicks'] + if click_limit is not None and get_user_monthly_clicks(conn, link['user_id']) >= click_limit: + return _visitor_page('Link temporarily unavailable', + 'This link has reached its monthly click limit. Please try again later.', + status=429) + return None + + @app.route('/') def redirect_link(code): if code in ('static', 'api', 'favicon.ico', 'robots.txt', 'sitemap.xml'): @@ -1820,27 +1970,37 @@ def redirect_link(code): with get_db() as conn: link = conn.execute('SELECT * FROM links WHERE code=? AND is_active=1', (code,)).fetchone() if not link: - return redirect('/?error=not_found') - if link['expires_at'] and link['expires_at'] < datetime.now(timezone.utc).replace(tzinfo=None).isoformat(): - return redirect('/?error=expired') + return _visitor_page('Link not found', + "This short link doesn't exist or has been removed.", + status=404) + gate = _check_link_gates(conn, link) + if gate: + return gate + if link['password_hash']: + return _password_form(code) + _record_click(link) + # 302 by default so browsers don't permanently cache the hop — + # keeps analytics accurate and destination edits effective. 301 is per-link opt-in. + return redirect(link['long_url'], code=link['redirect_type'] or 302) - # Monthly click quota check (skip for links with no owner) - if link['user_id']: - plan = get_user_plan(conn, link['user_id']) - click_limit = PLAN_LIMITS[plan]['monthly_clicks'] - if click_limit is not None and get_user_monthly_clicks(conn, link['user_id']) >= click_limit: - return redirect('/?error=quota_exceeded') - country = get_country_for_request() - client_ip = get_client_ip() - conn.execute( - 'INSERT INTO clicks (link_id,clicked_at,referrer,user_agent,ip_address,country) VALUES (?,?,?,?,?,?)', - (link['id'], datetime.now(timezone.utc).replace(tzinfo=None).isoformat(), - request.referrer, request.headers.get('User-Agent','')[:500], - client_ip[:45], country) - ) - conn.execute('UPDATE links SET clicks=clicks+1 WHERE id=?', (link['id'],)) - return redirect(link['long_url'], code=301) +@app.route('//unlock', methods=['POST']) +def unlock_link(code): + with get_db() as conn: + link = conn.execute('SELECT * FROM links WHERE code=? AND is_active=1', (code,)).fetchone() + if not link or not link['password_hash']: + return _visitor_page('Link not found', + "This short link doesn't exist or has been removed.", + status=404) + gate = _check_link_gates(conn, link) + if gate: + return gate + password = request.form.get('password', '') + if not check_password_hash(link['password_hash'], password): + return _password_form(code, error='Incorrect password — try again.') + _record_click(link) + # Always 302 for password links — the unlock must never be cached away + return redirect(link['long_url'], code=302) # ───────────────────────────────────────────── diff --git a/index.html b/index.html index 512b59a..7a0d517 100644 --- a/index.html +++ b/index.html @@ -1349,6 +1349,7 @@ function buildLinkItem(l) {
${ownerBadge} + ${l.has_password?'🔒':''}
${l.clicks} clicks${expNote}
@@ -1379,6 +1380,19 @@ function buildLinkItem(l) {
+
+ +
+
+
+ + ${l.has_password?``:''} +
+
@@ -1390,6 +1404,9 @@ function buildLinkItem(l) {
CLICK ANALYTICS
+
@@ -1469,8 +1486,12 @@ async function saveEdit(code) { const url = document.getElementById(`edit-url-${code}`).value.trim(); const title = document.getElementById(`edit-title-${code}`).value.trim(); const expiry = document.getElementById(`edit-expiry-${code}`).value; + const rt = parseInt(document.getElementById(`edit-rt-${code}`).value); + const pw = document.getElementById(`edit-pw-${code}`).value; const tags = document.getElementById(`edit-tags-${code}`).value.split(',').map(t=>t.trim()).filter(Boolean); - const resp = await authFetch(`${BASE}/api/links/${code}`, {method:'PATCH',body:JSON.stringify({url,title,expires_at:expiry||null,tags})}); + const body = {url, title, expires_at: expiry||null, tags, redirect_type: rt}; + if (pw) body.password = pw; // omit = leave password unchanged + const resp = await authFetch(`${BASE}/api/links/${code}`, {method:'PATCH',body:JSON.stringify(body)}); const data = await resp.json(); if (!resp.ok) { toast(data.error||'Save failed','error'); return; } toast('Link updated!'); @@ -1481,6 +1502,18 @@ async function saveEdit(code) { loadStats(); loadTags(); } +async function removeLinkPassword(code) { + if (!confirm(`Remove password protection from /${code}?`)) return; + const resp = await authFetch(`${BASE}/api/links/${code}`, {method:'PATCH', body:JSON.stringify({password:''})}); + const data = await resp.json(); + if (!resp.ok) { toast(data.error||'Failed','error'); return; } + toast('Password removed'); + const old = document.getElementById(`item-${code}`); + const fresh = buildLinkItem(data); + fresh.classList.add('expanded'); + old.parentNode.replaceChild(fresh, old); +} + async function deleteLink(code) { if (!confirm(`Delete /${code}?`)) return; await authFetch(`${BASE}/api/links/${code}`, {method:'DELETE'});