diff --git a/link_composer/__init__.py b/link_composer/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/link_composer/alphabets.py b/link_composer/alphabets.py new file mode 100644 index 0000000..7c16638 --- /dev/null +++ b/link_composer/alphabets.py @@ -0,0 +1 @@ +from .constants import ALPHABET_NAMES, PAYLOAD_ALPHABET, QR_ALPHABET, SEGMENT_ALPHABETS diff --git a/link_composer/base_n.py b/link_composer/base_n.py new file mode 100644 index 0000000..d0d1dc4 --- /dev/null +++ b/link_composer/base_n.py @@ -0,0 +1,28 @@ +def encode_int(value: int, alphabet: str) -> str: + if value < 0: + raise ValueError("cannot encode negative integers") + if value == 0: + return alphabet[0] + + base = len(alphabet) + out = [] + while value: + value, digit = divmod(value, base) + out.append(alphabet[digit]) + return "".join(reversed(out)) + +def decode_int(text: str, alphabet: str) -> int: + index = {char: i for i, char in enumerate(alphabet)} + value = 0 + base = len(alphabet) + for char in text: + if char not in index: + raise ValueError(f"invalid character for alphabet: {char!r}") + value = value * base + index[char] + return value + +def bytes_to_int(data: bytes) -> int: + return int.from_bytes(data, "big") if data else 0 + +def int_to_bytes(value: int, length: int) -> bytes: + return value.to_bytes(length, "big") if length else b"" diff --git a/link_composer/bitstream.py b/link_composer/bitstream.py new file mode 100644 index 0000000..d03d193 --- /dev/null +++ b/link_composer/bitstream.py @@ -0,0 +1,48 @@ +class BitWriter: + def __init__(self) -> None: + self._bits: list[int] = [] + + @property + def bit_length(self) -> int: + return len(self._bits) + + def write(self, value: int, width: int) -> None: + if width < 0: + raise ValueError("width must be positive") + if value < 0 or value >= (1 << width): + raise ValueError(f"value {value} does not fit in {width} bits") + for shift in range(width - 1, -1, -1): + self._bits.append((value >> shift) & 1) + + def to_bytes(self) -> bytes: + padding = (-len(self._bits)) % 8 + bits = self._bits + [0] * padding + out = bytearray() + for i in range(0, len(bits), 8): + byte = 0 + for bit in bits[i : i + 8]: + byte = (byte << 1) | bit + out.append(byte) + return bytes(out) + +class BitReader: + def __init__(self, data: bytes, bit_length: int) -> None: + self._bits: list[int] = [] + for byte in data: + for shift in range(7, -1, -1): + self._bits.append((byte >> shift) & 1) + self._bits = self._bits[:bit_length] + self._pos = 0 + + def read(self, width: int) -> int: + if self._pos + width > len(self._bits): + raise ValueError("payload ended unexpectedly") + value = 0 + for bit in self._bits[self._pos : self._pos + width]: + value = (value << 1) | bit + self._pos += width + return value + + @property + def remaining(self) -> int: + return len(self._bits) - self._pos diff --git a/link_composer/codec.py b/link_composer/codec.py new file mode 100644 index 0000000..8ca6943 --- /dev/null +++ b/link_composer/codec.py @@ -0,0 +1,221 @@ +from dataclasses import dataclass +from urllib.parse import urlsplit, urlunsplit +import re +import zlib + +from .constants import ( + ALPHABET_NAMES, + CODE_PREFIX_PAYLOAD, + CODE_PREFIX_QR, + CODE_SCHEMES, + CODE_VERSION, + DOMAIN_CODE_WIDTH, + LEGACY_DICTIONARY_CODE_WIDTH, + COMMON_DOMAINS, + COMMON_TLDS, + MODE_QR_STRUCTURED, + MODE_STRUCTURED, + PAYLOAD_ALPHABET, + QR_ALPHABET, + SCHEME_CODES, + SUPPORTED_CODE_VERSIONS, + SEGMENT_ALPHABETS, + SEPARATORS, + TLD_CODE_WIDTH, +) +from .base_n import bytes_to_int, decode_int, encode_int, int_to_bytes +from .bitstream import BitReader, BitWriter + +@dataclass(frozen=True) +class EncodeResult: + code: str + original_length: int + code_length: int + bit_length: int + mode: str + +def encode_url(url: str, *, qr: bool = False) -> EncodeResult: + normalized = _normalize_url(url) + writer = BitWriter() + writer.write(CODE_VERSION, 4) + + writer.write(0, 1) + _write_structured_url(writer, normalized) + + if qr: + alphabet = QR_ALPHABET + mode = MODE_QR_STRUCTURED + else: + alphabet = PAYLOAD_ALPHABET + mode = MODE_STRUCTURED + + bit_length = writer.bit_length + data = writer.to_bytes() + payload = encode_int(bytes_to_int(data), alphabet) + prefix = CODE_PREFIX_QR if qr else CODE_PREFIX_PAYLOAD + return EncodeResult( + code=prefix + payload, + original_length=len(normalized), + code_length=len(payload), + bit_length=bit_length, + mode=mode, + ) + +def decode_url(code: str) -> str: + if len(code) < 2: + raise ValueError("code is too short") + prefix, payload = code[0], code[1:] + if prefix == CODE_PREFIX_QR: + alphabet = QR_ALPHABET + elif prefix == CODE_PREFIX_PAYLOAD: + alphabet = PAYLOAD_ALPHABET + else: + raise ValueError("unknown code prefix") + value = decode_int(payload, alphabet) + byte_length = max(1, (value.bit_length() + 7) // 8) + data = int_to_bytes(value, byte_length) + + # We don't know the exact bit length from base conversion, so parse with all possible leading zero counts. + for leading_zero_bits in range(8): + candidate = (b"\x00" * (1 if leading_zero_bits else 0)) + data + bit_length = len(candidate) * 8 - leading_zero_bits + try: + return _decode_from_bytes(candidate, bit_length) + except ValueError: + continue + raise ValueError("invalid or unsupported code") + +def _decode_from_bytes(data: bytes, bit_length: int) -> str: + reader = BitReader(data, bit_length) + version = reader.read(4) + if version not in SUPPORTED_CODE_VERSIONS: + raise ValueError("unsupported version") + qr_mode = reader.read(1) + if qr_mode: + return _read_raw(reader) + return _read_structured_url(reader, version) + +def _normalize_url(url: str) -> str: + candidate = url.strip() + if not candidate: + raise ValueError("URL is empty") + if "://" not in candidate: + candidate = "https://" + candidate + parts = urlsplit(candidate) + if parts.scheme not in SCHEME_CODES: + raise ValueError("only http and https URLs are supported") + if not parts.netloc: + raise ValueError("URL requires a host") + return urlunsplit((parts.scheme.lower(), parts.netloc.lower(), parts.path, parts.query, parts.fragment)) + +def _write_structured_url(writer: BitWriter, url: str) -> None: + parts = urlsplit(url) + writer.write(SCHEME_CODES[parts.scheme], 1) + + host = parts.hostname or "" + has_www = host.startswith("www.") + if has_www: + host = host[4:] + writer.write(1 if has_www else 0, 1) + + labels = host.split(".") if host else [] + tld = labels[-1] if labels else "" + sld = labels[-2] if len(labels) >= 2 else "" + rest = ".".join(labels[:-2]) if len(labels) > 2 else "" + + _write_common_or_segment(writer, tld, COMMON_TLDS, TLD_CODE_WIDTH) + _write_common_or_segment(writer, sld, COMMON_DOMAINS, DOMAIN_CODE_WIDTH) + _write_text_segment(writer, rest) + + port = parts.port or 0 + writer.write(1 if port else 0, 1) + if port: + writer.write(port, 16) + + _write_text_segment(writer, parts.path) + _write_text_segment(writer, parts.query) + _write_text_segment(writer, parts.fragment) + +def _read_structured_url(reader: BitReader, version: int) -> str: + scheme = CODE_SCHEMES[reader.read(1)] + has_www = bool(reader.read(1)) + dictionary_width = LEGACY_DICTIONARY_CODE_WIDTH if version == 1 else TLD_CODE_WIDTH + tld = _read_common_or_segment(reader, COMMON_TLDS, dictionary_width) + sld = _read_common_or_segment(reader, COMMON_DOMAINS, dictionary_width) + rest = _read_text_segment(reader) + + labels = [part for part in [rest, sld] if part] + host = ".".join(labels + ([tld] if tld else [])) + if has_www: + host = "www." + host + + port = reader.read(16) if reader.read(1) else 0 + netloc = f"{host}:{port}" if port else host + path = _read_text_segment(reader) + query = _read_text_segment(reader) + fragment = _read_text_segment(reader) + return urlunsplit((scheme, netloc, path, query, fragment)) + +def _write_common_or_segment(writer: BitWriter, text: str, common: list[str], width: int) -> None: + if text in common[1:]: + writer.write(common.index(text), width) + else: + writer.write(0, width) + _write_text_segment(writer, text) + +def _read_common_or_segment(reader: BitReader, common: list[str], width: int) -> str: + code = reader.read(width) + if code: + if code >= len(common): + raise ValueError("invalid common dictionary code") + return common[code] + return _read_text_segment(reader) + +def _write_text_segment(writer: BitWriter, text: str) -> None: + if text == "": + writer.write(0, 1) + return + writer.write(1, 1) + + compressed = _compress_text(text) + raw = text.encode() + use_compressed = len(compressed) + 1 < len(raw) + writer.write(1 if use_compressed else 0, 1) + data = compressed if use_compressed else raw + writer.write(len(data), 16) + for byte in data: + writer.write(byte, 8) + +def _read_text_segment(reader: BitReader) -> str: + if not reader.read(1): + return "" + compressed = bool(reader.read(1)) + length = reader.read(16) + data = bytes(reader.read(8) for _ in range(length)) + if compressed: + data = zlib.decompress(data) + return data.decode() + +def _compress_text(text: str) -> bytes: + return zlib.compress(text.encode(), level=9) + +def _write_raw(writer: BitWriter, text: str) -> None: + data = text.encode() + writer.write(len(data), 16) + for byte in data: + writer.write(byte, 8) + +def _read_raw(reader: BitReader) -> str: + length = reader.read(16) + return bytes(reader.read(8) for _ in range(length)).decode() + +def describe_segments(url: str) -> list[dict[str, str | int]]: + normalized = _normalize_url(url) + chunks = [chunk for chunk in re.split(f"([{re.escape(SEPARATORS)}])", normalized) if chunk] + return [_describe_segment(chunk) for chunk in chunks] + +def _describe_segment(text: str) -> dict[str, str | int]: + for key, alphabet in SEGMENT_ALPHABETS.items(): + if alphabet is not None and all(char in alphabet for char in text): + return {"text": text, "alphabet": ALPHABET_NAMES[key], "size": len(alphabet)} + return {"text": text, "alphabet": ALPHABET_NAMES[7], "size": 256} diff --git a/link_composer/constants.py b/link_composer/constants.py new file mode 100644 index 0000000..40a92fe --- /dev/null +++ b/link_composer/constants.py @@ -0,0 +1,59 @@ +PAYLOAD_ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz-._~!$&'()*+,;=:@" +QR_ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ $%*+-./:" + +SEGMENT_ALPHABETS = { + 0: "0123456789", + 1: "abcdefghijklmnopqrstuvwxyz", + 2: "ABCDEFGHIJKLMNOPQRSTUVWXYZ", + 3: "abcdefghijklmnopqrstuvwxyz0123456789-", + 4: "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_", + 5: "0123456789ABCDEFabcdef", + 6: "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~!$&'()*+,;=:@", + 7: None, # raw UTF-8 bytes fallback +} + +ALPHABET_NAMES = { + 0: "decimal", + 1: "lowercase", + 2: "uppercase", + 3: "domain-like", + 4: "base64url-like", + 5: "hex", + 6: "url-safe", + 7: "utf8-raw", +} + +CODE_VERSION = 2 +SUPPORTED_CODE_VERSIONS = {1, 2} +TLD_CODE_WIDTH = 6 +DOMAIN_CODE_WIDTH = 6 +LEGACY_DICTIONARY_CODE_WIDTH = 4 + +CODE_PREFIX_PAYLOAD = "u" +CODE_PREFIX_QR = "Q" +MODE_STRUCTURED = "structured" +MODE_QR_STRUCTURED = "qr-structured" +MODE_DECODED = "decoded" + +SCHEME_CODES = {"http": 0, "https": 1} +CODE_SCHEMES = {value: key for key, value in SCHEME_CODES.items()} +ALLOWED_REDIRECT_SCHEMES = {"http", "https"} + +COMMON_TLDS = [ + "", "com", "org", "net", "io", "dev", "at", "de", "edu", "gov", "co", "uk", + "app", "ai", "me", "info", "biz", "xyz", "online", "site", "shop", "store", + "blog", "tech", "cloud", "digital", "tools", "software", "systems", "network", + "eu", "us", "ca", "au", "ch", "fr", "it", "es", "nl", "se", "no", "fi", + "dk", "pl", "cz", "jp", "cn", "in", "br", "ru", "tv", "fm", "to", "gg", +] +COMMON_DOMAINS = [ + "", "google", "youtube", "github", "reddit", "wikipedia", "twitter", "x", + "facebook", "instagram", "amazon", "microsoft", "apple", "cloudflare", "stackoverflow", + "linkedin", "tiktok", "whatsapp", "telegram", "discord", "slack", "notion", "medium", + "wordpress", "openai", "anthropic", "claude", "chatgpt", "docker", "kubernetes", "terraform", + "githubusercontent", "gitlab", "bitbucket", "npmjs", "pypi", "python", "nodejs", "mozilla", + "cloudfront", "aws", "azure", "googleapis", "gstatic", "vercel", "netlify", "heroku", + "paypal", "stripe", "shopify", "ebay", "etsy", "twitch", "spotify", "netflix", + "adobe", "figma", "canva", "atlassian", "jira", "confluence", "zoom", "dropbox", +] +SEPARATORS = "/?&=#.:-_~%+" diff --git a/link_composer/qr.py b/link_composer/qr.py new file mode 100644 index 0000000..219c3ec --- /dev/null +++ b/link_composer/qr.py @@ -0,0 +1,19 @@ +from io import BytesIO + +import qrcode +from qrcode.constants import ERROR_CORRECT_M + +def make_qr_png(data: str) -> bytes: + qr = qrcode.QRCode( + version=None, + error_correction=ERROR_CORRECT_M, + box_size=10, + border=4, + ) + qr.add_data(data, optimize=20) + qr.make(fit=True) + image = qr.make_image(fill_color="black", back_color="white") + + buffer = BytesIO() + image.save(buffer, format="PNG") + return buffer.getvalue() diff --git a/link_composer/security.py b/link_composer/security.py new file mode 100644 index 0000000..d76c693 --- /dev/null +++ b/link_composer/security.py @@ -0,0 +1,11 @@ +from urllib.parse import urlsplit + +from .constants import ALLOWED_REDIRECT_SCHEMES + +def validate_redirect_url(url: str) -> str: + parts = urlsplit(url) + if parts.scheme not in ALLOWED_REDIRECT_SCHEMES: + raise ValueError("decoded URL uses an unsupported redirect scheme") + if not parts.netloc: + raise ValueError("decoded URL has no host") + return url diff --git a/pyproject.toml b/pyproject.toml index afd2891..ff19e91 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "nanoshare" -version = "1.26.0" +version = "1.27.0" description = "Add your description here" readme = "README.md" requires-python = ">=3.13" @@ -16,6 +16,7 @@ dependencies = [ "quart==0.20.0", "quart-flask-patch==0.3.0", "quart-session", + "qrcode[pil]>=8.0", "redis==7.4.0", ] diff --git a/requirements.txt b/requirements.txt index 2cd808e..9dad42d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -41,6 +41,7 @@ pycparser==3.0 pygments==2.19.2 pyjwt==2.12.1 python-dotenv==1.2.2 +qrcode[pil]>=8.0 quart==0.20.0 quart-flask-patch==0.3.0 -e ./quart-session diff --git a/routes/__init__.py b/routes/__init__.py index d7f424a..d3c6702 100644 --- a/routes/__init__.py +++ b/routes/__init__.py @@ -18,6 +18,7 @@ auth_login_bp = create_auth_login_blueprint( from .side.main import side_main_bp from .side.upload import upload_bp +from .side.link_composer import link_composer_bp from .api.cli_auth import cli_auth_bp from .api.download import api_download_bp diff --git a/routes/side/link_composer.py b/routes/side/link_composer.py new file mode 100644 index 0000000..00c44b7 --- /dev/null +++ b/routes/side/link_composer.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +from urllib.parse import quote, unquote, urlparse + +from quart import Blueprint, Response, abort, jsonify, render_template, request, session, url_for + +from link_composer.codec import decode_url, encode_url +from link_composer.constants import ALLOWED_REDIRECT_SCHEMES +from link_composer.qr import make_qr_png +from link_composer.security import validate_redirect_url +from my_modules.app.setup import LIMITER +from my_modules.decoratory.header import login_required +from quart_common.web.wide_event import add_wide_event_context + +link_composer_bp = Blueprint("link_composer", __name__) + +LINK_REDIRECT_PREFIX = "/l/" +LINK_QR_PREFIX = "/links/qr/" + +def _external_url(endpoint: str, **values: str) -> str: + return request.url_root.rstrip("/") + url_for(endpoint, **values) + +def _external_code_url(prefix: str, code: str) -> str: + return request.url_root.rstrip("/") + prefix + quote(code, safe="") + +def _extract_code(value: str) -> str: + raw_value = value.strip() + parsed = urlparse(raw_value) + + if parsed.scheme in ALLOWED_REDIRECT_SCHEMES: + path = parsed.path + (f";{parsed.params}" if parsed.params else "") + if path.startswith(LINK_REDIRECT_PREFIX): + return unquote(path.removeprefix(LINK_REDIRECT_PREFIX)) + if path.startswith(LINK_QR_PREFIX): + return unquote(path.removeprefix(LINK_QR_PREFIX)) + + return raw_value + +def _decode_or_400(code: str) -> str: + try: + return validate_redirect_url(decode_url(_extract_code(code))) + except ValueError: + abort(400) + +@link_composer_bp.get("/links") +@LIMITER.limit("30 per minute;500 per hour;") +async def link_composer_page(): + add_wide_event_context(nanoshare={"operation": "link_composer_page"}) + return await render_template( + "views/webpage/links/composer.htm", + can_encode=session.get("user") is not None, + encode_api_url=url_for("link_composer.encode_link_api"), + decode_api_url=url_for("link_composer.decode_link_api"), + redirect_prefix=LINK_REDIRECT_PREFIX, + qr_prefix=LINK_QR_PREFIX, + ) + +@link_composer_bp.post("/api/links/encode") +@LIMITER.limit("30 per minute;500 per hour;") +@login_required +async def encode_link_api(user): + add_wide_event_context(nanoshare={"operation": "link_encode", "user_id": user.get("sub")}) + payload = await request.get_json(silent=True) or {} + url = str(payload.get("url") or "").strip() + qr_mode = bool(payload.get("qr")) + + try: + result = encode_url(url, qr=qr_mode) + decoded_url = validate_redirect_url(decode_url(result.code)) + except ValueError as exc: + return jsonify({"ok": False, "error": str(exc)}), 400 + + redirect_url = _external_code_url(LINK_REDIRECT_PREFIX, result.code) + qr_url = _external_code_url(LINK_QR_PREFIX, result.code) + + return jsonify({ + "ok": True, + "code": result.code, + "url": decoded_url, + "redirect_url": redirect_url, + "qr_url": qr_url, + "original_length": result.original_length, + "code_length": result.code_length, + "bit_length": result.bit_length, + "mode": result.mode, + "ratio": round(result.code_length / result.original_length, 3), + }) + +@link_composer_bp.post("/api/links/decode") +@LIMITER.limit("60 per minute;1000 per hour;") +async def decode_link_api(): + add_wide_event_context(nanoshare={"operation": "link_decode"}) + payload = await request.get_json(silent=True) or {} + code = _extract_code(str(payload.get("code") or "")) + + try: + url = validate_redirect_url(decode_url(code)) + except ValueError as exc: + return jsonify({"ok": False, "error": str(exc)}), 400 + + return jsonify({"ok": True, "url": url}) + +@link_composer_bp.get(f"{LINK_REDIRECT_PREFIX}") +@LIMITER.limit("120 per minute;2000 per hour;") +async def redirect_link(code: str): + add_wide_event_context(nanoshare={"operation": "link_redirect"}) + url = _decode_or_400(code) + return Response(status=307, headers={"Location": url, "Cache-Control": "no-store"}) + +@link_composer_bp.get(f"{LINK_QR_PREFIX}") +@LIMITER.limit("60 per minute;1000 per hour;") +async def qr_link(code: str): + add_wide_event_context(nanoshare={"operation": "link_qr"}) + _decode_or_400(code) + redirect_url = _external_code_url(LINK_REDIRECT_PREFIX, code) + return Response(make_qr_png(redirect_url), content_type="image/png") diff --git a/run.py b/run.py index 2292bab..330a905 100755 --- a/run.py +++ b/run.py @@ -11,6 +11,7 @@ from routes import ( basic_bp, auth_login_bp, side_main_bp, upload_bp, + link_composer_bp, cli_auth_bp, api_download_bp, api_events_bp, @@ -24,6 +25,7 @@ app.register_blueprint(auth_login_bp) app.register_blueprint(side_main_bp) app.register_blueprint(upload_bp) +app.register_blueprint(link_composer_bp) app.register_blueprint(cli_auth_bp) app.register_blueprint(api_download_bp) app.register_blueprint(api_events_bp) diff --git a/templates/side/elements/navigation.htm b/templates/side/elements/navigation.htm index 61b98c7..fd8170a 100644 --- a/templates/side/elements/navigation.htm +++ b/templates/side/elements/navigation.htm @@ -9,6 +9,8 @@