Compare commits

...
7 Commits
Author SHA1 Message Date
daniel156161 3194e534e0 fix(upload): preserve fields in multipart fallback
Build and Push Docker Container / build-and-push (push) Successful in 1m40s
- Recover regular multipart fields when the file part is not exposed through request.files.

- Preserve expires and note values for browser uploads that need body-based file recovery.

- Add parser coverage for file parts without filenames and recovered expiry fields.

- Extend upload route tests to assert recovered uploads keep expiry metadata.

- Bump NanoShare to 1.27.3.
2026-09-10 00:18:57 +02:00
daniel156161 21aca1b9d7 fix(upload): detect unnamed multipart files
Build and Push Docker Container / build-and-push (push) Successful in 1m38s
- Treat an existing multipart file field as upload content even when FileStorage has an empty filename.

- Keep upload telemetry aligned with file field presence instead of FileStorage truthiness.

- Add regression coverage for frontend-style uploads that provide a file field without a filename.

- Bump NanoShare to 1.27.2 and refresh the staged lockfile.
2026-09-09 23:42:30 +02:00
daniel156161 74ecc9b4b6 build: pull quart-flask-patch from gitea registry
Build and Push Docker Container / build-and-push (push) Successful in 1m9s
- Resolve quart-flask-patch from the private Gitea PyPI index
- Require quart-flask-patch >= 0.3.1, the version published there
- Drop the PyPI 0.3.0 build that crashed the server on Python 3.14
- The Gitea build declares requires-python >=3.13,<3.15
- Relock uv.lock so sdist and wheel point at the Gitea URLs
- Bump NanoShare to 1.27.1

Verified with uv lock --check, uv sync --frozen and an import of
quart_flask_patch reporting 0.3.1. Test suite stays at 141 passed.
2026-09-06 03:06:51 +02:00
daniel156161 772a178c11 fix(observability): bump quart_common and refresh stale error tests
Build and Push Docker Container / build-and-push (push) Successful in 1m35s
- Bump quart_common to 53b7599 with the httpx request-less crash fix
- The global exception handler no longer dies on RequestError without request
- Expect 503 for Convex connect failures, 504 is reserved for timeouts
- Add an API timeout case so the 504 branch stays covered
- Register a link_composer blueprint so error.htm can render its navigation
- Move the legacy logger assertion into a WIDE_EVENT_ENABLED=false test
- Assert user.session_age_seconds instead of the removed session.age_seconds
- Expect no legacy middleware logs while wide-event logging is active

The middleware moved into quart_common's register_security_middleware, so the
NanoShare copies of those assertions had been describing pre-refactor
behaviour. Suite is green again at 141 passed.
2026-09-06 02:57:37 +02:00
daniel156161 2279633a81 build: pull quart-session from gitea registry, bump to python 3.14
Build and Push Docker Container / build-and-push (push) Successful in 1m12s
- Drop the quart-session git submodule and its uv workspace member entry
- Resolve quart-session from the private Gitea PyPI index instead
- Pin the new index as explicit so only quart-session is fetched from it
- Require quart-session >= 3.0.2, the version published in the registry
- Raise requires-python to >=3.14 and set .python-version to 3.14
- Switch the image base to ghcr.io/astral-sh/uv:python3.14-trixie-slim
- Relax runtime pins from == to >= so uv.lock stays the single source
- Relock all 50 packages for CPython 3.14, dropping 3.13-only markers
- Delete the stale hand-maintained requirements.txt, unused by the build

Verified with uv lock --check, uv sync --frozen and a docker build; the
container reports CPython 3.14.7 and imports quart_session and convex.
Test suite is unchanged at 131 passed, 4 pre-existing wide_event failures.
2026-09-06 02:45:22 +02:00
daniel156161 ed62837f6d feat: add self-contained link composer
Build and Push Docker Container / build-and-push (push) Successful in 1m36s
- Add structured URL codec with compact alphabets and dictionary support.
- Add public decode, redirect, and QR endpoints for composed links.
- Add authenticated encode API and link composer UI.
- Generate PNG QR codes via qrcode with Pillow support.
- Register the new blueprint and navigation entry.
- Cover public decode, auth gating, redirect URL parsing, and QR output.
- Bump NanoShare to 1.27.0 and lock new dependencies.
2026-08-14 08:56:42 +02:00
daniel156161 d3e5e11f8d fix(cli): ignore generated folders in subfolders
Build and Push Docker Container / build-and-push (push) Successful in 3m12s
- Match directory ignore patterns against every path segment.
- Prevent nested .comments folders from uploading XML sidecar files.
- Bump the standalone CLI to 0.4.2 for the ignore fix.
2026-07-28 00:14:00 +02:00
32 changed files with 1628 additions and 543 deletions
-3
View File
@@ -1,6 +1,3 @@
[submodule "quart-session"]
path = quart-session
url = git@git.yiprawr.dev:daniel156161/quart-session.git
[submodule "my_helpers"] [submodule "my_helpers"]
path = my_helpers path = my_helpers
url = git@git.yiprawr.dev:daniel156161/python-helper-modules.git url = git@git.yiprawr.dev:daniel156161/python-helper-modules.git
+1 -1
View File
@@ -1 +1 @@
3.13 3.14
+1 -1
View File
@@ -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 # Install app
COPY . /app COPY . /app
+2
View File
@@ -46,6 +46,8 @@ def is_ignored(rel_path: str, patterns: list[str]) -> bool:
if pattern.endswith('/'): if pattern.endswith('/'):
if path == normalized or path.startswith(f'{normalized}/'): if path == normalized or path.startswith(f'{normalized}/'):
return True return True
if any(fnmatch(part, normalized) for part in parts):
return True
continue continue
if '/' in normalized: if '/' in normalized:
if fnmatch(path, normalized): if fnmatch(path, normalized):
+1 -1
View File
@@ -1,6 +1,6 @@
[project] [project]
name = "nanoshare-cli" name = "nanoshare-cli"
version = "0.4.1" version = "0.4.2"
description = "NanoShare desktop CLI and folder sync client" description = "NanoShare desktop CLI and folder sync client"
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" requires-python = ">=3.11"
View File
+1
View File
@@ -0,0 +1 @@
from .constants import ALPHABET_NAMES, PAYLOAD_ALPHABET, QR_ALPHABET, SEGMENT_ALPHABETS
+28
View File
@@ -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""
+48
View File
@@ -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
+221
View File
@@ -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}
+59
View File
@@ -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 = "/?&=#.:-_~%+"
+19
View File
@@ -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()
+11
View File
@@ -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
+71
View File
@@ -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
View File
@@ -1,22 +1,23 @@
[project] [project]
name = "nanoshare" name = "nanoshare"
version = "1.26.0" version = "1.27.3"
description = "Add your description here" description = "Add your description here"
readme = "README.md" readme = "README.md"
requires-python = ">=3.13" requires-python = ">=3.14"
dependencies = [ dependencies = [
"aiohttp==3.13.3", "aiohttp>=3.13.3",
"aiologger==0.7.0", "aiologger>=0.7.0",
"convex==0.7.0", "convex>=0.7.0",
"flask-limiter==4.1.1", "flask-limiter>=4.1.1",
"httpx==0.28.1", "httpx>=0.28.1",
"hypercorn==0.18.0", "hypercorn>=0.18.0",
"joserfc>=1.6.5", "joserfc>=1.6.5",
"python-dotenv==1.2.2", "python-dotenv>=1.2.2",
"quart==0.20.0", "quart>=0.20.0",
"quart-flask-patch==0.3.0", "quart-flask-patch>=0.3.1",
"quart-session", "quart-session>=3.0.2",
"redis==7.4.0", "qrcode[pil]>=8.0",
"redis>=7.4.0",
] ]
[tool.pytest.ini_options] [tool.pytest.ini_options]
@@ -25,10 +26,11 @@ testpaths = [
"quart_common/tests", "quart_common/tests",
] ]
[tool.uv.workspace] [[tool.uv.index]]
members = [ name = "gitea"
"quart-session", url = "https://git.yiprawr.dev/api/packages/daniel156161/pypi/simple/"
] explicit = true
[tool.uv.sources] [tool.uv.sources]
quart-session = { workspace = true } quart-flask-patch = { index = "gitea" }
quart-session = { index = "gitea" }
Submodule quart-session deleted from ce173e43c7
-55
View File
@@ -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
+1
View File
@@ -18,6 +18,7 @@ auth_login_bp = create_auth_login_blueprint(
from .side.main import side_main_bp from .side.main import side_main_bp
from .side.upload import upload_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.cli_auth import cli_auth_bp
from .api.download import api_download_bp from .api.download import api_download_bp
+116
View File
@@ -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
View File
@@ -1,6 +1,7 @@
from my_modules.decoratory.header import login_required from my_modules.decoratory.header import login_required
from my_modules.expiry import parse_expires, ensure_utc from my_modules.expiry import parse_expires, ensure_utc
from my_modules.file_meta import iso_stamp_filename, format_size 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_common.web.wide_event import add_wide_event_context
from quart import Blueprint, request, jsonify, current_app from quart import Blueprint, request, jsonify, current_app
@@ -74,19 +75,37 @@ async def api_upload(user):
text = form.get('text', '') text = form.get('text', '')
orphan_registry = getattr(current_app, 'orphan_storage_registry', None) 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') 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)) 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"}) add_wide_event_context(nanoshare={"operation_status": "missing_content"})
return jsonify({'ok': False, 'error': 'No content provided'}), 400 return jsonify({'ok': False, 'error': 'No content provided'}), 400
content_type = None content_type = None
# --- binary upload path --- # --- binary upload path ---
if uploaded: if has_uploaded_file:
fname = uploaded.filename or '' fname = uploaded.filename or ''
add_wide_event_context(nanoshare={"upload_type": "file", "filename_present": bool(fname)}) add_wide_event_context(nanoshare={"upload_type": "file", "filename_present": bool(fname)})
ctype = uploaded.mimetype or 'application/octet-stream' ctype = uploaded.mimetype or 'application/octet-stream'
+2
View File
@@ -11,6 +11,7 @@ from routes import (
basic_bp, auth_login_bp, basic_bp, auth_login_bp,
side_main_bp, side_main_bp,
upload_bp, upload_bp,
link_composer_bp,
cli_auth_bp, cli_auth_bp,
api_download_bp, api_download_bp,
api_events_bp, api_events_bp,
@@ -24,6 +25,7 @@ app.register_blueprint(auth_login_bp)
app.register_blueprint(side_main_bp) app.register_blueprint(side_main_bp)
app.register_blueprint(upload_bp) app.register_blueprint(upload_bp)
app.register_blueprint(link_composer_bp)
app.register_blueprint(cli_auth_bp) app.register_blueprint(cli_auth_bp)
app.register_blueprint(api_download_bp) app.register_blueprint(api_download_bp)
app.register_blueprint(api_events_bp) app.register_blueprint(api_events_bp)
+2
View File
@@ -9,6 +9,8 @@
<ul id="nav-list"> <ul id="nav-list">
<li><a href="{{ url_for('side_main.index') }}">NanoShare</a></li> <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 %} {% if user %}
<li><a href="{{ url_for('side_main.index') }}">Upload</a></li> <li><a href="{{ url_for('side_main.index') }}">Upload</a></li>
<li><a href="{{ url_for('side_main.files_list') }}">Files</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">&nbsp;</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 %}
+44 -3
View File
@@ -48,9 +48,14 @@
} }
/* Globale Stile */ /* Globale Stile */
html, body { html {
height: -webkit-fill-available; min-height: 100%;
height: 100%; background: var(--bg);
}
body {
min-height: 100vh;
min-height: 100dvh;
display: flex; display: flex;
flex-direction: column; flex-direction: column;
font-family: Arial, sans-serif; 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(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%), radial-gradient(900px 500px at 110% 10%, rgba(142,246,255,.18), transparent 60%),
var(--bg); 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"; 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} .toast.show{display:flex;animation:pop .18s ease-out}
@keyframes pop{from{transform:translateY(8px);opacity:0}to{transform:translateY(0);opacity:1}} @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 ====== */ /* ====== Utilities ====== */
.sr-only{ .sr-only{
position:absolute !important;width:1px;height:1px;padding:0;margin:-1px; position:absolute !important;width:1px;height:1px;padding:0;margin:-1px;
+39 -1
View File
@@ -77,6 +77,10 @@ def register_template_routes(app):
auth_login.add_url_rule("/logout", "logout", index) auth_login.add_url_rule("/logout", "logout", index)
app.register_blueprint(auth_login) 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): def load_errors_and_basics(monkeypatch, app):
logger = FakeLogger() 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") 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) assert any("Name or service not known" in str(error) for error in logger.errors)
asyncio.run(run_test()) 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") response = await app.test_client().get("/api/failing")
payload = await response.get_json() 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 response.status_code == 504
assert payload["error"] == "Database Error" assert payload["error"] == "Database Error"
+126
View File
@@ -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())
+3
View File
@@ -213,6 +213,8 @@ def test_sync_ignores_local_files_from_nanoshareignore(tmp_path, monkeypatch):
(tmp_path / 'scratch.tmp').write_text('ignored') (tmp_path / 'scratch.tmp').write_text('ignored')
(tmp_path / '.comments').mkdir() (tmp_path / '.comments').mkdir()
(tmp_path / '.comments' / 'ignored.xml').write_text('<comment />') (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 / '.env').write_text('SECRET=ignored')
(tmp_path / '.venv').mkdir() (tmp_path / '.venv').mkdir()
(tmp_path / '.venv' / 'ignored.py').write_text('ignored') (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) patterns = load_ignore_patterns(tmp_path)
assert is_ignored('.comments/comment.xml', patterns) 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', patterns)
assert is_ignored('.env.local', patterns) assert is_ignored('.env.local', patterns)
assert is_ignored('.venv/bin/python', patterns) assert is_ignored('.venv/bin/python', patterns)
+28
View File
@@ -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'
+146
View File
@@ -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())
+3 -2
View File
@@ -98,7 +98,8 @@ def test_middleware_adds_user_and_client_context(monkeypatch):
assert "authenticated" not in g.wide_event["session"] assert "authenticated" not in g.wide_event["session"]
assert g.wide_event["session"]["permanent"] is True assert g.wide_event["session"]["permanent"] is True
assert g.wide_event["session"]["login_at_unix"] == 1 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" assert g.wide_event["client"]["ip"] == "203.0.113.10"
asyncio.run(run_test()) 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["client"]["ip"] == "203.0.113.12"
assert g.wide_event["security"] == {"ip_lookup_failed": True} assert g.wide_event["security"] == {"ip_lookup_failed": True}
assert g.wide_event["error"]["type"] == "RuntimeError" 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()) asyncio.run(run_test())
Generated
+393 -452
View File
File diff suppressed because it is too large Load Diff