Compare commits
7
Commits
08d0d154c9
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3194e534e0
|
||
|
|
21aca1b9d7
|
||
|
|
74ecc9b4b6
|
||
|
|
772a178c11
|
||
|
|
2279633a81
|
||
|
|
ed62837f6d
|
||
|
|
d3e5e11f8d
|
@@ -1,6 +1,3 @@
|
||||
[submodule "quart-session"]
|
||||
path = quart-session
|
||||
url = git@git.yiprawr.dev:daniel156161/quart-session.git
|
||||
[submodule "my_helpers"]
|
||||
path = my_helpers
|
||||
url = git@git.yiprawr.dev:daniel156161/python-helper-modules.git
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
3.13
|
||||
3.14
|
||||
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
FROM ghcr.io/astral-sh/uv:python3.13-trixie-slim
|
||||
FROM ghcr.io/astral-sh/uv:python3.14-trixie-slim
|
||||
|
||||
# Install app
|
||||
COPY . /app
|
||||
|
||||
@@ -46,6 +46,8 @@ def is_ignored(rel_path: str, patterns: list[str]) -> bool:
|
||||
if pattern.endswith('/'):
|
||||
if path == normalized or path.startswith(f'{normalized}/'):
|
||||
return True
|
||||
if any(fnmatch(part, normalized) for part in parts):
|
||||
return True
|
||||
continue
|
||||
if '/' in normalized:
|
||||
if fnmatch(path, normalized):
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "nanoshare-cli"
|
||||
version = "0.4.1"
|
||||
version = "0.4.2"
|
||||
description = "NanoShare desktop CLI and folder sync client"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.11"
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
from .constants import ALPHABET_NAMES, PAYLOAD_ALPHABET, QR_ALPHABET, SEGMENT_ALPHABETS
|
||||
@@ -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""
|
||||
@@ -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
|
||||
@@ -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}
|
||||
@@ -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 = "/?&=#.:-_~%+"
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -0,0 +1,71 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from io import BytesIO
|
||||
|
||||
from quart.datastructures import FileStorage
|
||||
from werkzeug.datastructures import Headers
|
||||
from werkzeug.http import parse_options_header
|
||||
from werkzeug.sansio.multipart import Data, Epilogue, Field, File, NeedData, MultipartDecoder
|
||||
|
||||
@dataclass
|
||||
class MultipartUploadParts:
|
||||
fields: dict[str, str] = field(default_factory=dict)
|
||||
file: FileStorage | None = None
|
||||
|
||||
def _content_type(headers: Headers) -> str:
|
||||
return headers.get('content-type') or 'application/octet-stream'
|
||||
|
||||
def _fallback_filename(headers: Headers, default: str = 'file') -> str:
|
||||
disposition = headers.get('content-disposition', '')
|
||||
_, options = parse_options_header(disposition)
|
||||
filename = options.get('filename')
|
||||
return filename if filename is not None else default
|
||||
|
||||
def parse_multipart_upload_body(body: bytes, boundary: str | bytes | None, file_field_name: str = 'file') -> MultipartUploadParts:
|
||||
parts = MultipartUploadParts()
|
||||
if not body or not boundary:
|
||||
return parts
|
||||
|
||||
boundary_bytes = boundary.encode() if isinstance(boundary, str) else boundary
|
||||
parser = MultipartDecoder(boundary_bytes)
|
||||
parser.receive_data(body)
|
||||
parser.receive_data(None)
|
||||
|
||||
current_part = None
|
||||
current_chunks: list[bytes] = []
|
||||
|
||||
while True:
|
||||
event = parser.next_event()
|
||||
if isinstance(event, (Epilogue, NeedData)):
|
||||
break
|
||||
if isinstance(event, (Field, File)):
|
||||
current_part = event
|
||||
current_chunks = []
|
||||
continue
|
||||
if not isinstance(event, Data) or current_part is None:
|
||||
continue
|
||||
|
||||
current_chunks.append(event.data)
|
||||
if event.more_data:
|
||||
continue
|
||||
|
||||
data = b''.join(current_chunks)
|
||||
if current_part.name == file_field_name and data:
|
||||
headers = current_part.headers
|
||||
parts.file = FileStorage(
|
||||
stream=BytesIO(data),
|
||||
filename=getattr(current_part, 'filename', None) or _fallback_filename(headers),
|
||||
name=file_field_name,
|
||||
content_type=_content_type(headers),
|
||||
headers=headers,
|
||||
)
|
||||
elif isinstance(current_part, Field):
|
||||
_, options = parse_options_header(current_part.headers.get('content-type', ''))
|
||||
charset = options.get('charset') or 'utf-8'
|
||||
parts.fields[current_part.name] = data.decode(charset, 'replace')
|
||||
|
||||
return parts
|
||||
|
||||
def recover_file_from_multipart_body(body: bytes, boundary: str | bytes | None, field_name: str = 'file') -> FileStorage | None:
|
||||
return parse_multipart_upload_body(body, boundary, field_name).file
|
||||
+20
-18
@@ -1,22 +1,23 @@
|
||||
[project]
|
||||
name = "nanoshare"
|
||||
version = "1.26.0"
|
||||
version = "1.27.3"
|
||||
description = "Add your description here"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.13"
|
||||
requires-python = ">=3.14"
|
||||
dependencies = [
|
||||
"aiohttp==3.13.3",
|
||||
"aiologger==0.7.0",
|
||||
"convex==0.7.0",
|
||||
"flask-limiter==4.1.1",
|
||||
"httpx==0.28.1",
|
||||
"hypercorn==0.18.0",
|
||||
"aiohttp>=3.13.3",
|
||||
"aiologger>=0.7.0",
|
||||
"convex>=0.7.0",
|
||||
"flask-limiter>=4.1.1",
|
||||
"httpx>=0.28.1",
|
||||
"hypercorn>=0.18.0",
|
||||
"joserfc>=1.6.5",
|
||||
"python-dotenv==1.2.2",
|
||||
"quart==0.20.0",
|
||||
"quart-flask-patch==0.3.0",
|
||||
"quart-session",
|
||||
"redis==7.4.0",
|
||||
"python-dotenv>=1.2.2",
|
||||
"quart>=0.20.0",
|
||||
"quart-flask-patch>=0.3.1",
|
||||
"quart-session>=3.0.2",
|
||||
"qrcode[pil]>=8.0",
|
||||
"redis>=7.4.0",
|
||||
]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
@@ -25,10 +26,11 @@ testpaths = [
|
||||
"quart_common/tests",
|
||||
]
|
||||
|
||||
[tool.uv.workspace]
|
||||
members = [
|
||||
"quart-session",
|
||||
]
|
||||
[[tool.uv.index]]
|
||||
name = "gitea"
|
||||
url = "https://git.yiprawr.dev/api/packages/daniel156161/pypi/simple/"
|
||||
explicit = true
|
||||
|
||||
[tool.uv.sources]
|
||||
quart-session = { workspace = true }
|
||||
quart-flask-patch = { index = "gitea" }
|
||||
quart-session = { index = "gitea" }
|
||||
|
||||
Submodule quart-session deleted from ce173e43c7
+1
-1
Submodule quart_common updated: d61514d755...53b7599193
@@ -1,55 +0,0 @@
|
||||
aiofiles==25.1.0
|
||||
aiohappyeyeballs==2.6.1
|
||||
aiohttp==3.13.3
|
||||
aiologger==0.7.0
|
||||
aiosignal==1.4.0
|
||||
anyio==4.13.0
|
||||
attrs==26.1.0
|
||||
authlib==1.6.9
|
||||
blinker==1.9.0
|
||||
certifi==2026.2.25
|
||||
cffi==2.0.0
|
||||
click==8.3.1
|
||||
convex==0.7.0
|
||||
cryptography==46.0.6
|
||||
deprecated==1.3.1
|
||||
dotenv==0.9.9
|
||||
flask==3.1.3
|
||||
flask-limiter==4.1.1
|
||||
frozenlist==1.8.0
|
||||
h11==0.16.0
|
||||
h2==4.3.0
|
||||
hpack==4.1.0
|
||||
httpcore==1.0.9
|
||||
httpx==0.28.1
|
||||
hypercorn==0.18.0
|
||||
hyperframe==6.1.0
|
||||
idna==3.11
|
||||
itsdangerous==2.2.0
|
||||
jinja2==3.1.6
|
||||
limits==5.8.0
|
||||
markdown-it-py==4.0.0
|
||||
markupsafe==3.0.3
|
||||
mdurl==0.1.2
|
||||
msgpack==1.1.2
|
||||
multidict==6.7.1
|
||||
ordered-set==4.1.0
|
||||
packaging==26.0
|
||||
priority==2.0.0
|
||||
propcache==0.4.1
|
||||
pycparser==3.0
|
||||
pygments==2.19.2
|
||||
pyjwt==2.12.1
|
||||
python-dotenv==1.2.2
|
||||
quart==0.20.0
|
||||
quart-flask-patch==0.3.0
|
||||
-e ./quart-session
|
||||
redis==7.4.0
|
||||
rich==14.3.3
|
||||
setuptools==82.0.1
|
||||
sniffio==1.3.1
|
||||
typing-extensions==4.15.0
|
||||
werkzeug==3.1.7
|
||||
wrapt==2.1.2
|
||||
wsproto==1.3.2
|
||||
yarl==1.23.0
|
||||
@@ -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
|
||||
|
||||
@@ -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}<path:code>")
|
||||
@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}<path:code>")
|
||||
@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")
|
||||
+23
-4
@@ -1,6 +1,7 @@
|
||||
from my_modules.decoratory.header import login_required
|
||||
from my_modules.expiry import parse_expires, ensure_utc
|
||||
from my_modules.file_meta import iso_stamp_filename, format_size
|
||||
from my_modules.upload_content import parse_multipart_upload_body
|
||||
from quart_common.web.wide_event import add_wide_event_context
|
||||
|
||||
from quart import Blueprint, request, jsonify, current_app
|
||||
@@ -74,19 +75,37 @@ async def api_upload(user):
|
||||
text = form.get('text', '')
|
||||
orphan_registry = getattr(current_app, 'orphan_storage_registry', None)
|
||||
|
||||
add_wide_event_context(nanoshare={"operation": "upload", "has_file": bool(files.get('file')), "has_text": bool(text.strip())})
|
||||
|
||||
uploaded = files.get('file')
|
||||
recovered_upload = False
|
||||
recovered_fields = {}
|
||||
if uploaded is None and request.mimetype == 'multipart/form-data':
|
||||
recovered_parts = parse_multipart_upload_body(
|
||||
await request.get_data(cache=True),
|
||||
request.mimetype_params.get('boundary'),
|
||||
)
|
||||
uploaded = recovered_parts.file
|
||||
recovered_fields = recovered_parts.fields
|
||||
recovered_upload = uploaded is not None
|
||||
|
||||
if recovered_fields:
|
||||
note = recovered_fields.get('note', note)
|
||||
expires_raw = recovered_fields.get('expires', expires_raw)
|
||||
text = recovered_fields.get('text', text)
|
||||
|
||||
has_uploaded_file = uploaded is not None
|
||||
|
||||
add_wide_event_context(nanoshare={"operation": "upload", "has_file": has_uploaded_file, "has_text": bool(text.strip()), "recovered_file": recovered_upload, "recovered_field_count": len(recovered_fields)})
|
||||
|
||||
expires_at_dt = ensure_utc(parse_expires(expires_raw))
|
||||
|
||||
if not uploaded and not text.strip():
|
||||
if not has_uploaded_file and not text.strip():
|
||||
add_wide_event_context(nanoshare={"operation_status": "missing_content"})
|
||||
return jsonify({'ok': False, 'error': 'No content provided'}), 400
|
||||
|
||||
content_type = None
|
||||
|
||||
# --- binary upload path ---
|
||||
if uploaded:
|
||||
if has_uploaded_file:
|
||||
fname = uploaded.filename or ''
|
||||
add_wide_event_context(nanoshare={"upload_type": "file", "filename_present": bool(fname)})
|
||||
ctype = uploaded.mimetype or 'application/octet-stream'
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -9,6 +9,8 @@
|
||||
<ul id="nav-list">
|
||||
<li><a href="{{ url_for('side_main.index') }}">NanoShare</a></li>
|
||||
|
||||
<li><a href="{{ url_for('link_composer.link_composer_page') }}">Links</a></li>
|
||||
|
||||
{% if user %}
|
||||
<li><a href="{{ url_for('side_main.index') }}">Upload</a></li>
|
||||
<li><a href="{{ url_for('side_main.files_list') }}">Files</a></li>
|
||||
|
||||
@@ -0,0 +1,218 @@
|
||||
{% extends "base.htm" %}
|
||||
|
||||
{% block title %}NanoShare - Link Composer{% endblock %}
|
||||
|
||||
{% block meta %}
|
||||
<meta name="description" content="Encode, decode and QR-share self-contained NanoShare links.">
|
||||
<meta name="robots" content="noindex, nofollow" />
|
||||
{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<main class="wrap link-composer">
|
||||
<section class="card hero-card">
|
||||
<span class="hero-badge"><i class="fa-solid fa-link"></i> Link Composer</span>
|
||||
<h1 class="title">Encode, decode and QR links</h1>
|
||||
<p class="subtitle">Decode is public. Log in to create compressed links and QR codes.</p>
|
||||
</section>
|
||||
|
||||
<section class="composer-grid {% if not can_encode %}is-single{% endif %}">
|
||||
{% if can_encode %}
|
||||
<section class="card composer-card">
|
||||
<div class="card-heading">
|
||||
<h2>Encode</h2>
|
||||
</div>
|
||||
|
||||
<label for="compose-url">URL</label>
|
||||
<textarea id="compose-url" placeholder="https://example.com/docs?page=1"></textarea>
|
||||
<label class="switch-row">
|
||||
<input id="qr-mode" type="checkbox">
|
||||
QR alphabet mode
|
||||
</label>
|
||||
<div class="actions-row">
|
||||
<button id="encode-btn" class="btn" type="button">Encode URL</button>
|
||||
<button id="clear-btn" class="btn btn-ghost" type="button">Clear</button>
|
||||
</div>
|
||||
<div id="encode-error" class="form-error"></div>
|
||||
</section>
|
||||
{% endif %}
|
||||
|
||||
<section class="card composer-card">
|
||||
<div class="card-heading">
|
||||
<h2>Decode</h2>
|
||||
</div>
|
||||
<label for="compose-code">Code or redirect link</label>
|
||||
<textarea id="compose-code" placeholder="u... or https://example.com/l/u..."></textarea>
|
||||
{% if can_encode %}
|
||||
<div class="switch-row switch-row-spacer" aria-hidden="true"> </div>
|
||||
{% endif %}
|
||||
<div class="actions-row">
|
||||
<button id="decode-btn" class="btn" type="button">Decode Code</button>
|
||||
{% if not can_encode %}
|
||||
<a class="btn btn-ghost" href="{{ url_for('auth_login.login') }}">Login to encode</a>
|
||||
{% endif %}
|
||||
</div>
|
||||
<div id="decode-error" class="form-error"></div>
|
||||
</section>
|
||||
</section>
|
||||
|
||||
<section id="result-panel" class="card result-panel">
|
||||
<div class="card-heading">
|
||||
<h2>Result</h2>
|
||||
<span id="mode-badge" class="badge">structured</span>
|
||||
</div>
|
||||
|
||||
<div class="result-grid">
|
||||
<div class="result-main">
|
||||
<div class="output-block">
|
||||
<span class="output-label">Compressed code</span>
|
||||
<div class="copy-field">
|
||||
<code id="out-code"></code>
|
||||
<button class="btn btn-ghost mini" type="button" data-copy="out-code">Copy</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="output-block">
|
||||
<span class="output-label">Redirect link</span>
|
||||
<div class="copy-field">
|
||||
<code id="out-redirect"></code>
|
||||
<button class="btn btn-ghost mini" type="button" data-copy="out-redirect">Copy</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="output-block">
|
||||
<span class="output-label">Decoded URL</span>
|
||||
<div class="copy-field">
|
||||
<code id="out-url"></code>
|
||||
<button class="btn btn-ghost mini" type="button" data-copy="out-url">Copy</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="stats-grid">
|
||||
<div class="stat-card"><strong id="stat-original">–</strong><span>Original</span></div>
|
||||
<div class="stat-card"><strong id="stat-code">–</strong><span>Code</span></div>
|
||||
<div class="stat-card"><strong id="stat-ratio">–</strong><span>Ratio</span></div>
|
||||
<div class="stat-card"><strong id="stat-bits">–</strong><span>Bits</span></div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="qr-card">
|
||||
<img id="qr-image" alt="Generated QR code">
|
||||
<div class="actions-row centered">
|
||||
<a id="qr-open" class="btn btn-ghost" target="_blank" rel="noreferrer">Open PNG</a>
|
||||
<a id="qr-download" class="btn" download="nanoshare-link-qr.png">Download PNG</a>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</section>
|
||||
</main>
|
||||
{% endblock %}
|
||||
|
||||
{% block script %}
|
||||
<script>
|
||||
const canEncode = {{ 'true' if can_encode else 'false' }};
|
||||
const encodeApiUrl = {{ encode_api_url | tojson }};
|
||||
const decodeApiUrl = {{ decode_api_url | tojson }};
|
||||
const redirectPrefix = {{ redirect_prefix | tojson }};
|
||||
const qrPrefix = {{ qr_prefix | tojson }};
|
||||
const $ = (id) => document.getElementById(id);
|
||||
|
||||
function setError(id, message) {
|
||||
const el = $(id);
|
||||
el.textContent = message || '';
|
||||
el.classList.toggle('visible', Boolean(message));
|
||||
}
|
||||
|
||||
function setText(id, value) {
|
||||
$(id).textContent = value || '';
|
||||
}
|
||||
|
||||
async function postJson(path, body) {
|
||||
const response = await fetch(path, {
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
const data = await response.json();
|
||||
if (!response.ok || data.ok === false) throw new Error(data.error || data.detail || 'Request failed');
|
||||
return data;
|
||||
}
|
||||
|
||||
function showResult(data) {
|
||||
setText('out-code', data.code || $('compose-code').value.trim());
|
||||
setText('out-url', data.url);
|
||||
setText('out-redirect', data.redirect_url || `${location.origin}${redirectPrefix}${encodeURIComponent(data.code || $('compose-code').value.trim())}`);
|
||||
setText('stat-original', data.original_length ?? '–');
|
||||
setText('stat-code', data.code_length ?? ((data.code || '').length || '–'));
|
||||
setText('stat-ratio', data.ratio ?? '–');
|
||||
setText('stat-bits', data.bit_length ?? '–');
|
||||
setText('mode-badge', data.mode || 'decoded');
|
||||
|
||||
const code = data.code || $('compose-code').value.trim();
|
||||
if (code) {
|
||||
const qrUrl = data.qr_url || `${location.origin}${qrPrefix}${encodeURIComponent(code)}`;
|
||||
$('qr-image').src = qrUrl;
|
||||
$('qr-open').href = qrUrl;
|
||||
$('qr-download').href = qrUrl;
|
||||
}
|
||||
$('result-panel').classList.add('visible');
|
||||
}
|
||||
|
||||
$('encode-btn')?.addEventListener('click', async () => {
|
||||
if (!canEncode) return;
|
||||
setError('encode-error', '');
|
||||
try {
|
||||
const data = await postJson(encodeApiUrl, { url: $('compose-url').value, qr: $('qr-mode').checked });
|
||||
$('compose-code').value = data.code;
|
||||
showResult(data);
|
||||
} catch (error) {
|
||||
setError('encode-error', error.message);
|
||||
}
|
||||
});
|
||||
|
||||
function extractCode(value) {
|
||||
const trimmed = value.trim();
|
||||
try {
|
||||
const url = new URL(trimmed);
|
||||
const fullPath = url.pathname + url.search + url.hash;
|
||||
if (fullPath.startsWith(redirectPrefix)) return decodeURIComponent(fullPath.slice(redirectPrefix.length));
|
||||
if (fullPath.startsWith(qrPrefix)) return decodeURIComponent(fullPath.slice(qrPrefix.length));
|
||||
} catch (_) {}
|
||||
return trimmed;
|
||||
}
|
||||
|
||||
$('decode-btn').addEventListener('click', async () => {
|
||||
setError('decode-error', '');
|
||||
try {
|
||||
const input = $('compose-code').value.trim();
|
||||
const code = extractCode(input);
|
||||
const data = await postJson(decodeApiUrl, { code: input });
|
||||
showResult({
|
||||
...data,
|
||||
code,
|
||||
redirect_url: `${location.origin}${redirectPrefix}${encodeURIComponent(code)}`,
|
||||
qr_url: `${location.origin}${qrPrefix}${encodeURIComponent(code)}`,
|
||||
mode: 'decoded',
|
||||
});
|
||||
} catch (error) {
|
||||
setError('decode-error', error.message);
|
||||
}
|
||||
});
|
||||
|
||||
$('clear-btn')?.addEventListener('click', () => {
|
||||
$('compose-url').value = '';
|
||||
$('compose-code').value = '';
|
||||
$('result-panel').classList.remove('visible');
|
||||
setError('encode-error', '');
|
||||
setError('decode-error', '');
|
||||
});
|
||||
|
||||
document.addEventListener('click', async (event) => {
|
||||
const button = event.target.closest('[data-copy]');
|
||||
if (!button) return;
|
||||
await navigator.clipboard.writeText($(button.dataset.copy).textContent);
|
||||
const old = button.textContent;
|
||||
button.textContent = 'Copied';
|
||||
setTimeout(() => button.textContent = old, 900);
|
||||
});
|
||||
</script>
|
||||
{% endblock %}
|
||||
@@ -48,9 +48,14 @@
|
||||
}
|
||||
|
||||
/* Globale Stile */
|
||||
html, body {
|
||||
height: -webkit-fill-available;
|
||||
height: 100%;
|
||||
html {
|
||||
min-height: 100%;
|
||||
background: var(--bg);
|
||||
}
|
||||
|
||||
body {
|
||||
min-height: 100vh;
|
||||
min-height: 100dvh;
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
font-family: Arial, sans-serif;
|
||||
@@ -300,6 +305,8 @@ body{
|
||||
radial-gradient(1200px 600px at 10% -10%, rgba(124,158,255,.18), transparent 60%),
|
||||
radial-gradient(900px 500px at 110% 10%, rgba(142,246,255,.18), transparent 60%),
|
||||
var(--bg);
|
||||
background-attachment: fixed;
|
||||
background-repeat: no-repeat;
|
||||
font:500 14px/1.5 system-ui,-apple-system,Segoe UI,Inter,Roboto,"Helvetica Neue",Arial,"Noto Sans","Apple Color Emoji","Segoe UI Emoji";
|
||||
}
|
||||
|
||||
@@ -528,6 +535,40 @@ input[type="file"]{display:none}
|
||||
.toast.show{display:flex;animation:pop .18s ease-out}
|
||||
@keyframes pop{from{transform:translateY(8px);opacity:0}to{transform:translateY(0);opacity:1}}
|
||||
|
||||
/* ====== Link Composer ====== */
|
||||
.link-composer{max-width:1400px;width:min(1400px, calc(100% - 32px));padding:clamp(18px,2.6vw,34px)}
|
||||
.link-composer .hero-card{margin-bottom:22px;padding:clamp(24px,3vw,36px)}
|
||||
.link-composer .composer-card,.link-composer .result-panel{padding:clamp(24px,3vw,36px)}
|
||||
.composer-grid{display:grid;grid-template-columns:repeat(2,minmax(0,1fr));gap:24px;margin-bottom:24px}
|
||||
.composer-grid.is-single{grid-template-columns:1fr}
|
||||
@media (max-width:860px){.composer-grid{grid-template-columns:1fr}}
|
||||
.composer-card textarea{
|
||||
width:100%;height:190px;min-height:190px;resize:vertical;border-radius:14px;border:1px solid var(--border);
|
||||
background:color-mix(in srgb,var(--panel) 88%, transparent);color:var(--text);padding:16px 18px;
|
||||
}
|
||||
.composer-card textarea:focus{outline:none;border-color:color-mix(in srgb,var(--accent) 50%, var(--border));box-shadow:0 0 0 3px color-mix(in srgb,var(--accent) 25%, transparent)}
|
||||
.composer-card textarea:disabled,.composer-card button:disabled{opacity:.55;cursor:not-allowed}
|
||||
.card-heading{display:flex;justify-content:space-between;gap:12px;align-items:center;margin-bottom:14px}
|
||||
.card-heading h2{margin:0;font-size:1.25rem}
|
||||
.badge-warning{color:#ffd37a;background:rgba(255,211,122,.12);border-color:rgba(255,211,122,.35)}
|
||||
.notice{display:grid;gap:4px;margin:10px 0 16px;padding:12px;border:1px solid rgba(255,211,122,.35);border-radius:12px;background:rgba(255,211,122,.08);color:var(--muted)}
|
||||
.switch-row{display:flex;align-items:center;gap:10px;margin:12px 0;color:var(--muted);font-weight:700}
|
||||
.switch-row-spacer{visibility:hidden;pointer-events:none}
|
||||
.switch-row input{accent-color:var(--accent)}
|
||||
.actions-row{display:flex;gap:12px;flex-wrap:wrap;margin-top:18px}.actions-row.centered{justify-content:center}
|
||||
.form-error{display:none;margin-top:12px;padding:12px;border-radius:12px;border:1px solid rgba(255,122,122,.45);background:rgba(255,122,122,.10);color:#ffd5d5}.form-error.visible{display:block}
|
||||
.result-panel{display:none}.result-panel.visible{display:block}
|
||||
.result-grid{display:grid;grid-template-columns:minmax(0,1fr) 320px;gap:24px;align-items:stretch}
|
||||
@media (max-width:820px){.result-grid{grid-template-columns:1fr}}
|
||||
.output-block{margin-top:12px}.output-label{display:block;color:var(--muted);font-size:.78rem;font-weight:800;text-transform:uppercase;letter-spacing:.08em;margin-bottom:7px}
|
||||
.copy-field{display:flex;gap:10px;align-items:center;justify-content:space-between;border:1px solid var(--border);border-radius:12px;background:rgba(0,0,0,.16);padding:10px 12px;overflow-wrap:anywhere}
|
||||
.copy-field code{font:700 .93rem ui-monospace,SFMono-Regular,Menlo,Consolas,monospace;color:var(--text)}
|
||||
.btn.mini{padding:.4rem .6rem;font-size:.82rem;flex:0 0 auto}
|
||||
.stats-grid{display:grid;grid-template-columns:repeat(4,minmax(0,1fr));gap:10px;margin-top:14px}
|
||||
@media (max-width:620px){.stats-grid{grid-template-columns:repeat(2,minmax(0,1fr))}}
|
||||
.stat-card{border:1px solid var(--border);border-radius:12px;background:rgba(0,0,0,.16);padding:12px}.stat-card strong{display:block;font-size:1.15rem}.stat-card span{color:var(--muted);font-size:.82rem}
|
||||
.qr-card{display:flex;flex-direction:column;gap:12px;justify-content:space-between;align-items:center;border:1px solid var(--border);border-radius:14px;background:rgba(0,0,0,.12);padding:14px}.qr-card img{max-width:230px;width:100%;background:#fff;border-radius:12px;padding:10px}
|
||||
|
||||
/* ====== Utilities ====== */
|
||||
.sr-only{
|
||||
position:absolute !important;width:1px;height:1px;padding:0;margin:-1px;
|
||||
|
||||
@@ -77,6 +77,10 @@ def register_template_routes(app):
|
||||
auth_login.add_url_rule("/logout", "logout", index)
|
||||
app.register_blueprint(auth_login)
|
||||
|
||||
link_composer = Blueprint("link_composer", __name__)
|
||||
link_composer.add_url_rule("/links", "link_composer_page", index)
|
||||
app.register_blueprint(link_composer)
|
||||
|
||||
|
||||
def load_errors_and_basics(monkeypatch, app):
|
||||
logger = FakeLogger()
|
||||
@@ -96,7 +100,24 @@ def test_convex_connect_error_is_returned_as_global_database_error(monkeypatch):
|
||||
|
||||
response = await app.test_client().get("/favicon.ico")
|
||||
|
||||
assert response.status_code == 504
|
||||
assert response.status_code == 503
|
||||
assert response.headers["Retry-After"]
|
||||
assert logger.errors == []
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
def test_convex_connect_error_uses_legacy_logger_when_wide_events_are_disabled(monkeypatch):
|
||||
async def run_test():
|
||||
monkeypatch.setenv("WIDE_EVENT_ENABLED", "false")
|
||||
app = Quart(__name__, template_folder=str(Path(__file__).resolve().parents[1] / "templates" / "side"))
|
||||
register_template_routes(app)
|
||||
app.convex = FailingConvex()
|
||||
_errors, basics, logger = load_errors_and_basics(monkeypatch, app)
|
||||
app.register_blueprint(basics.basic_bp)
|
||||
|
||||
response = await app.test_client().get("/favicon.ico")
|
||||
|
||||
assert response.status_code == 503
|
||||
assert any("Name or service not known" in str(error) for error in logger.errors)
|
||||
|
||||
asyncio.run(run_test())
|
||||
@@ -113,6 +134,23 @@ def test_api_convex_connect_error_returns_json_database_error(monkeypatch):
|
||||
response = await app.test_client().get("/api/failing")
|
||||
payload = await response.get_json()
|
||||
|
||||
assert response.status_code == 503
|
||||
assert payload["error"] == "Database Error"
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
def test_api_convex_timeout_returns_gateway_timeout(monkeypatch):
|
||||
async def run_test():
|
||||
app = Quart(__name__, template_folder=str(Path(__file__).resolve().parents[1] / "templates" / "side"))
|
||||
load_errors_and_basics(monkeypatch, app)
|
||||
|
||||
@app.get("/api/slow")
|
||||
async def slow_api():
|
||||
raise httpx.ReadTimeout("convex took too long")
|
||||
|
||||
response = await app.test_client().get("/api/slow")
|
||||
payload = await response.get_json()
|
||||
|
||||
assert response.status_code == 504
|
||||
assert payload["error"] == "Database Error"
|
||||
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
import asyncio
|
||||
import importlib.util
|
||||
import sys
|
||||
import types
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote
|
||||
|
||||
from quart import Quart, g, session
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
|
||||
class _NoOpLimiter:
|
||||
def limit(self, *_args, **_kwargs):
|
||||
return lambda fn: fn
|
||||
|
||||
def _load_module(module_name, module_path):
|
||||
spec = importlib.util.spec_from_file_location(module_name, module_path)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[module_name] = module
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
def _install_stubs(monkeypatch):
|
||||
setup_stub = types.ModuleType("my_modules.app.setup")
|
||||
setup_stub.LIMITER = _NoOpLimiter()
|
||||
monkeypatch.setitem(sys.modules, "my_modules.app.setup", setup_stub)
|
||||
|
||||
header_stub = types.ModuleType("my_modules.decoratory.header")
|
||||
|
||||
def login_required(func):
|
||||
async def wrapper(*args, **kwargs):
|
||||
user = session.get("user")
|
||||
if user is None:
|
||||
return "unauthorized", 401
|
||||
return await func(user=user, *args, **kwargs)
|
||||
wrapper.__name__ = func.__name__
|
||||
return wrapper
|
||||
|
||||
header_stub.login_required = login_required
|
||||
monkeypatch.setitem(sys.modules, "my_modules.decoratory.header", header_stub)
|
||||
|
||||
def _client(monkeypatch):
|
||||
_install_stubs(monkeypatch)
|
||||
root = Path(__file__).resolve().parents[1]
|
||||
module = _load_module("link_composer_route_under_test", root / "routes" / "side" / "link_composer.py")
|
||||
app = Quart(__name__, template_folder=str(root / "templates" / "side"))
|
||||
app.secret_key = "test-secret"
|
||||
app.register_blueprint(module.link_composer_bp)
|
||||
|
||||
@app.before_request
|
||||
async def wide_event_context():
|
||||
g.wide_event = {}
|
||||
|
||||
return app, app.test_client()
|
||||
|
||||
def test_decode_api_is_public(monkeypatch):
|
||||
async def run_test():
|
||||
_app, client = _client(monkeypatch)
|
||||
from link_composer.codec import encode_url
|
||||
|
||||
code = encode_url("https://example.com/target").code
|
||||
response = await client.post("/api/links/decode", json={"code": code})
|
||||
payload = await response.get_json()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert payload["ok"] is True
|
||||
assert payload["url"] == "https://example.com/target"
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
def test_decode_api_accepts_redirect_url(monkeypatch):
|
||||
async def run_test():
|
||||
_app, client = _client(monkeypatch)
|
||||
from link_composer.codec import encode_url
|
||||
|
||||
code = encode_url("https://example.com/target").code
|
||||
response = await client.post("/api/links/decode", json={"code": f"http://localhost/l/{code}"})
|
||||
payload = await response.get_json()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert payload["ok"] is True
|
||||
assert payload["url"] == "https://example.com/target"
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
def test_encode_api_requires_login(monkeypatch):
|
||||
async def run_test():
|
||||
_app, client = _client(monkeypatch)
|
||||
|
||||
response = await client.post("/api/links/encode", json={"url": "https://example.com/target"})
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
def test_encode_api_returns_qr_for_logged_in_user(monkeypatch):
|
||||
async def run_test():
|
||||
_app, client = _client(monkeypatch)
|
||||
async with client.session_transaction() as sess:
|
||||
sess["user"] = {"sub": "user_123", "name": "Demo"}
|
||||
|
||||
response = await client.post("/api/links/encode", json={"url": "https://example.com/target"})
|
||||
payload = await response.get_json()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert payload["ok"] is True
|
||||
assert payload["code"].startswith("u")
|
||||
encoded_code = quote(payload["code"], safe="")
|
||||
assert payload["redirect_url"].endswith(f"/l/{encoded_code}")
|
||||
assert f"/links/qr/{encoded_code}" in payload["qr_url"]
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
def test_qr_endpoint_returns_png_after_decoding(monkeypatch):
|
||||
async def run_test():
|
||||
_app, client = _client(monkeypatch)
|
||||
from link_composer.codec import encode_url
|
||||
|
||||
code = encode_url("https://example.com/target").code
|
||||
response = await client.get(f"/links/qr/{code}")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.headers["Content-Type"] == "image/png"
|
||||
assert (await response.get_data()).startswith(b"\x89PNG")
|
||||
|
||||
asyncio.run(run_test())
|
||||
@@ -213,6 +213,8 @@ def test_sync_ignores_local_files_from_nanoshareignore(tmp_path, monkeypatch):
|
||||
(tmp_path / 'scratch.tmp').write_text('ignored')
|
||||
(tmp_path / '.comments').mkdir()
|
||||
(tmp_path / '.comments' / 'ignored.xml').write_text('<comment />')
|
||||
(tmp_path / 'Cerberus' / '.comments').mkdir(parents=True)
|
||||
(tmp_path / 'Cerberus' / '.comments' / '76. Two-headeds.PNG.xml').write_text('<comment />')
|
||||
(tmp_path / '.env').write_text('SECRET=ignored')
|
||||
(tmp_path / '.venv').mkdir()
|
||||
(tmp_path / '.venv' / 'ignored.py').write_text('ignored')
|
||||
@@ -261,6 +263,7 @@ def test_default_ignore_patterns_skip_common_generated_files(tmp_path):
|
||||
patterns = load_ignore_patterns(tmp_path)
|
||||
|
||||
assert is_ignored('.comments/comment.xml', patterns)
|
||||
assert is_ignored('Cerberus/.comments/76. Two-headeds.PNG.xml', patterns)
|
||||
assert is_ignored('.env', patterns)
|
||||
assert is_ignored('.env.local', patterns)
|
||||
assert is_ignored('.venv/bin/python', patterns)
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
from my_modules.upload_content import parse_multipart_upload_body, recover_file_from_multipart_body
|
||||
|
||||
def test_recovers_file_part_without_filename():
|
||||
boundary = '----nanoshare-test-boundary'
|
||||
body = (
|
||||
f'--{boundary}\r\n'
|
||||
'Content-Disposition: form-data; name="file"\r\n'
|
||||
'Content-Type: application/octet-stream\r\n'
|
||||
'\r\n'
|
||||
).encode() + b'hello upload' + (
|
||||
'\r\n'
|
||||
f'--{boundary}\r\n'
|
||||
'Content-Disposition: form-data; name="expires"\r\n'
|
||||
'\r\n'
|
||||
'7d\r\n'
|
||||
f'--{boundary}--\r\n'
|
||||
).encode()
|
||||
|
||||
uploaded = recover_file_from_multipart_body(body, boundary)
|
||||
parts = parse_multipart_upload_body(body, boundary)
|
||||
|
||||
assert uploaded is not None
|
||||
assert uploaded.filename == 'file'
|
||||
assert uploaded.mimetype == 'application/octet-stream'
|
||||
assert uploaded.stream.read() == b'hello upload'
|
||||
assert parts.fields == {'expires': '7d'}
|
||||
assert parts.file is not None
|
||||
assert parts.file.stream.read() == b'hello upload'
|
||||
@@ -0,0 +1,146 @@
|
||||
import asyncio
|
||||
import importlib.util
|
||||
import sys
|
||||
import types
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
from quart import Quart, session
|
||||
from quart.datastructures import FileStorage
|
||||
|
||||
def load_upload_module(monkeypatch):
|
||||
fake_header = types.ModuleType('my_modules.decoratory.header')
|
||||
|
||||
def login_required(func):
|
||||
async def wrapper(*args, **kwargs):
|
||||
return await func(user=session.get('user'), *args, **kwargs)
|
||||
return wrapper
|
||||
|
||||
fake_header.login_required = login_required
|
||||
|
||||
fake_event = types.ModuleType('quart_common.web.wide_event')
|
||||
fake_event.add_wide_event_context = lambda **kwargs: None
|
||||
|
||||
monkeypatch.setitem(sys.modules, 'my_modules.decoratory.header', fake_header)
|
||||
monkeypatch.setitem(sys.modules, 'quart_common.web.wide_event', fake_event)
|
||||
|
||||
module_path = Path(__file__).resolve().parents[1] / 'routes' / 'side' / 'upload.py'
|
||||
spec = importlib.util.spec_from_file_location('upload_route_under_test', module_path)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
class AwaitableValue:
|
||||
def __init__(self, value):
|
||||
self.value = value
|
||||
|
||||
def __await__(self):
|
||||
async def get_value():
|
||||
return self.value
|
||||
return get_value().__await__()
|
||||
|
||||
class FakeRequest:
|
||||
def __init__(self, form, files, body=b'', boundary=None):
|
||||
self.form = AwaitableValue(form)
|
||||
self.files = AwaitableValue(files)
|
||||
self.mimetype = 'multipart/form-data' if boundary else ''
|
||||
self.mimetype_params = {'boundary': boundary} if boundary else {}
|
||||
self._body = body
|
||||
|
||||
async def get_data(self, cache=True):
|
||||
return self._body
|
||||
|
||||
class FakeConvex:
|
||||
def __init__(self):
|
||||
self.sent = []
|
||||
self.files = []
|
||||
|
||||
async def send_to_storage(self, data, content_type):
|
||||
self.sent.append((data, content_type))
|
||||
return 'storage_1'
|
||||
|
||||
async def send_stream_to_storage(self, stream, content_type):
|
||||
data = stream.read()
|
||||
self.sent.append((data, content_type))
|
||||
return 'storage_1', len(data)
|
||||
|
||||
async def add_file(self, **kwargs):
|
||||
self.files.append(kwargs)
|
||||
return {'file_id': 'file_1'}
|
||||
|
||||
def test_upload_accepts_file_field_without_filename(monkeypatch):
|
||||
async def run_test():
|
||||
upload = load_upload_module(monkeypatch)
|
||||
|
||||
app = Quart(__name__)
|
||||
app.secret_key = 'test-secret'
|
||||
app.convex = FakeConvex()
|
||||
app.orphan_storage_registry = None
|
||||
|
||||
async with app.test_request_context('/api/upload', method='POST'):
|
||||
session['user'] = {'sub': 'user_1'}
|
||||
|
||||
uploaded = FileStorage(
|
||||
stream=BytesIO(b'hello'),
|
||||
filename='',
|
||||
name='file',
|
||||
content_type='application/octet-stream',
|
||||
)
|
||||
fake_request = FakeRequest(
|
||||
form={'expires': '7d', 'note': 'test', 'text': ''},
|
||||
files={'file': uploaded},
|
||||
)
|
||||
monkeypatch.setattr(upload, 'request', fake_request)
|
||||
|
||||
response = await upload.api_upload()
|
||||
|
||||
assert await response.get_json() == {'ok': True}
|
||||
assert app.convex.sent == [(b'hello', 'application/octet-stream')]
|
||||
assert app.convex.files[0]['file_name'].endswith('.bin')
|
||||
assert app.convex.files[0]['expires_at'] is not None
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
def test_upload_recovers_file_field_from_multipart_body(monkeypatch):
|
||||
async def run_test():
|
||||
upload = load_upload_module(monkeypatch)
|
||||
|
||||
app = Quart(__name__)
|
||||
app.secret_key = 'test-secret'
|
||||
app.convex = FakeConvex()
|
||||
app.orphan_storage_registry = None
|
||||
|
||||
boundary = '----nanoshare-test-boundary'
|
||||
body = (
|
||||
f'--{boundary}\r\n'
|
||||
'Content-Disposition: form-data; name="file"\r\n'
|
||||
'Content-Type: application/octet-stream\r\n'
|
||||
'\r\n'
|
||||
).encode() + b'hello recovered' + (
|
||||
'\r\n'
|
||||
f'--{boundary}\r\n'
|
||||
'Content-Disposition: form-data; name="expires"\r\n'
|
||||
'\r\n'
|
||||
'7d\r\n'
|
||||
f'--{boundary}--\r\n'
|
||||
).encode()
|
||||
|
||||
async with app.test_request_context('/api/upload', method='POST'):
|
||||
session['user'] = {'sub': 'user_1'}
|
||||
fake_request = FakeRequest(
|
||||
form={'expires': '7d', 'note': 'test', 'text': ''},
|
||||
files={},
|
||||
body=body,
|
||||
boundary=boundary,
|
||||
)
|
||||
monkeypatch.setattr(upload, 'request', fake_request)
|
||||
|
||||
response = await upload.api_upload()
|
||||
|
||||
assert await response.get_json() == {'ok': True}
|
||||
assert app.convex.sent == [(b'hello recovered', 'application/octet-stream')]
|
||||
assert app.convex.files[0]['file_name'].endswith('.bin')
|
||||
assert app.convex.files[0]['expires_at'] is not None
|
||||
assert app.convex.files[0]['note'] == 'test'
|
||||
|
||||
asyncio.run(run_test())
|
||||
@@ -98,7 +98,8 @@ def test_middleware_adds_user_and_client_context(monkeypatch):
|
||||
assert "authenticated" not in g.wide_event["session"]
|
||||
assert g.wide_event["session"]["permanent"] is True
|
||||
assert g.wide_event["session"]["login_at_unix"] == 1
|
||||
assert g.wide_event["session"]["age_seconds"] >= 0
|
||||
assert "age_seconds" not in g.wide_event["session"]
|
||||
assert g.wide_event["user"]["session_age_seconds"] >= 0
|
||||
assert g.wide_event["client"]["ip"] == "203.0.113.10"
|
||||
|
||||
asyncio.run(run_test())
|
||||
@@ -143,6 +144,6 @@ def test_middleware_records_convex_security_failures(monkeypatch):
|
||||
assert g.wide_event["client"]["ip"] == "203.0.113.12"
|
||||
assert g.wide_event["security"] == {"ip_lookup_failed": True}
|
||||
assert g.wide_event["error"]["type"] == "RuntimeError"
|
||||
assert fake_logger.records == [("error", "[MIDDLEWARE] Convex ip_lookup failed: convex down")]
|
||||
assert fake_logger.records == []
|
||||
|
||||
asyncio.run(run_test())
|
||||
|
||||
Reference in New Issue
Block a user