forked from retoor/devplacepy
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a8ed5b690f |
File diff suppressed because one or more lines are too long
+3
-11
@@ -15,27 +15,19 @@ def avatar_seed(user) -> str:
|
|||||||
return user.get("avatar_seed") or user.get("username") or ""
|
return user.get("avatar_seed") or user.get("username") or ""
|
||||||
|
|
||||||
|
|
||||||
def _svg_with_size(svg: str, size: int) -> str:
|
def generate_avatar_svg(seed: str) -> str:
|
||||||
size_attr = f' width="{size}" height="{size}"'
|
|
||||||
if svg.startswith("<svg"):
|
|
||||||
tag_end = svg.index(">")
|
|
||||||
return svg[:tag_end] + size_attr + svg[tag_end:]
|
|
||||||
return svg
|
|
||||||
|
|
||||||
|
|
||||||
def generate_avatar_svg(seed: str, size: int = 128) -> str:
|
|
||||||
try:
|
try:
|
||||||
from multiavatar.multiavatar import multiavatar
|
from multiavatar.multiavatar import multiavatar
|
||||||
|
|
||||||
svg = multiavatar(seed, None, None)
|
svg = multiavatar(seed, None, None)
|
||||||
if svg and svg.strip().startswith("<svg"):
|
if svg and svg.strip().startswith("<svg"):
|
||||||
return _svg_with_size(svg, size)
|
return svg
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.exception(e)
|
logger.exception(e)
|
||||||
logger.warning(f"Avatar generation failed for {seed}: {e}")
|
logger.warning(f"Avatar generation failed for {seed}: {e}")
|
||||||
initial = seed[:1].upper() if seed else "?"
|
initial = seed[:1].upper() if seed else "?"
|
||||||
return (
|
return (
|
||||||
f'<svg xmlns="http://www.w3.org/2000/svg" width="{size}" height="{size}" viewBox="0 0 100 100">'
|
'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 100 100">'
|
||||||
f'<rect width="100" height="100" rx="50" fill="#ff6b35"/>'
|
f'<rect width="100" height="100" rx="50" fill="#ff6b35"/>'
|
||||||
f'<text x="50" y="65" text-anchor="middle" fill="white" font-size="40" font-weight="700" font-family="sans-serif">{initial}</text></svg>'
|
f'<text x="50" y="65" text-anchor="middle" fill="white" font-size="40" font-weight="700" font-family="sans-serif">{initial}</text></svg>'
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -470,9 +470,19 @@ class SeoRunForm(BaseModel):
|
|||||||
text = value.strip()
|
text = value.strip()
|
||||||
if not text:
|
if not text:
|
||||||
raise ValueError("A URL is required")
|
raise ValueError("A URL is required")
|
||||||
|
if "://" in text:
|
||||||
|
scheme = text.split("://", 1)[0]
|
||||||
|
if scheme not in ("http", "https"):
|
||||||
|
raise ValueError(f"Only http and https URLs are allowed; got '{scheme}://'")
|
||||||
|
else:
|
||||||
|
text = f"https://{text}"
|
||||||
|
if not SEO_URL_PATTERN.match(text):
|
||||||
|
raise ValueError("URL must be a valid http or https source location")
|
||||||
return text
|
return text
|
||||||
|
|
||||||
|
|
||||||
|
SEO_URL_PATTERN = re.compile(r"^https?://[a-zA-Z0-9][\w./:@~^?&#%=;-]*$")
|
||||||
|
|
||||||
ISSLOP_URL_PATTERN = re.compile(r"^(https?://|git://|ssh://|git@)[\w./:@~^-]+$", re.IGNORECASE)
|
ISSLOP_URL_PATTERN = re.compile(r"^(https?://|git://|ssh://|git@)[\w./:@~^-]+$", re.IGNORECASE)
|
||||||
ISSLOP_SINGLE_SLASH_PATTERN = re.compile(r"^(https?|git|ssh):/(?!/)", re.IGNORECASE)
|
ISSLOP_SINGLE_SLASH_PATTERN = re.compile(r"^(https?|git|ssh):/(?!/)", re.IGNORECASE)
|
||||||
ISSLOP_SCHEME_PATTERN = re.compile(r"^[a-z][a-z0-9+.-]*://", re.IGNORECASE)
|
ISSLOP_SCHEME_PATTERN = re.compile(r"^[a-z][a-z0-9+.-]*://", re.IGNORECASE)
|
||||||
|
|||||||
@@ -17,16 +17,14 @@ _CACHE_CONTROL = f"public, max-age={SECONDS_PER_DAY}, immutable"
|
|||||||
|
|
||||||
@router.get("/{style}/{seed}")
|
@router.get("/{style}/{seed}")
|
||||||
async def avatar_proxy(request: Request, style: str, seed: str, size: int = 128):
|
async def avatar_proxy(request: Request, style: str, seed: str, size: int = 128):
|
||||||
size = max(16, min(512, size))
|
etag = '"' + hashlib.md5(f"{seed}:{size}".encode("utf-8")).hexdigest() + '"'
|
||||||
cache_key = f"{seed}:{size}"
|
|
||||||
etag = '"' + hashlib.md5(cache_key.encode("utf-8")).hexdigest() + '"'
|
|
||||||
headers = {"ETag": etag, "Cache-Control": _CACHE_CONTROL}
|
headers = {"ETag": etag, "Cache-Control": _CACHE_CONTROL}
|
||||||
|
|
||||||
if request.headers.get("if-none-match") == etag:
|
if request.headers.get("if-none-match") == etag:
|
||||||
return Response(status_code=304, headers=headers)
|
return Response(status_code=304, headers=headers)
|
||||||
|
|
||||||
svg = _cache.get(cache_key)
|
svg = _cache.get(seed)
|
||||||
if svg is None:
|
if svg is None:
|
||||||
svg = generate_avatar_svg(seed, size)
|
svg = generate_avatar_svg(seed)
|
||||||
_cache.set(cache_key, svg)
|
_cache.set(seed, svg)
|
||||||
return Response(content=svg, media_type="image/svg+xml", headers=headers)
|
return Response(content=svg, media_type="image/svg+xml", headers=headers)
|
||||||
|
|||||||
+1
-29
@@ -1,6 +1,7 @@
|
|||||||
# retoor <retoor@molodetz.nl>
|
# retoor <retoor@molodetz.nl>
|
||||||
|
|
||||||
import requests
|
import requests
|
||||||
|
from devplacepy.avatar import avatar_url, generate_avatar_svg
|
||||||
from tests.conftest import BASE_URL
|
from tests.conftest import BASE_URL
|
||||||
|
|
||||||
|
|
||||||
@@ -9,32 +10,3 @@ def test_avatar_endpoint_serves_svg(app_server):
|
|||||||
assert r.status_code == 200
|
assert r.status_code == 200
|
||||||
assert "svg" in r.headers.get("content-type", "").lower()
|
assert "svg" in r.headers.get("content-type", "").lower()
|
||||||
assert "<svg" in r.text
|
assert "<svg" in r.text
|
||||||
|
|
||||||
|
|
||||||
def test_avatar_respects_size(app_server):
|
|
||||||
r = requests.get(f"{BASE_URL}/avatar/multiavatar/alice_test?size=32")
|
|
||||||
assert r.status_code == 200
|
|
||||||
assert 'width="32"' in r.text
|
|
||||||
assert 'height="32"' in r.text
|
|
||||||
|
|
||||||
|
|
||||||
def test_avatar_different_sizes_different_response(app_server):
|
|
||||||
r32 = requests.get(f"{BASE_URL}/avatar/multiavatar/bob_test?size=32")
|
|
||||||
r128 = requests.get(f"{BASE_URL}/avatar/multiavatar/bob_test?size=128")
|
|
||||||
assert r32.status_code == 200
|
|
||||||
assert r128.status_code == 200
|
|
||||||
etag32 = r32.headers.get("etag", "")
|
|
||||||
etag128 = r128.headers.get("etag", "")
|
|
||||||
assert etag32 != etag128, "same seed with different sizes must produce different ETags"
|
|
||||||
|
|
||||||
|
|
||||||
def test_avatar_size_clamped_low(app_server):
|
|
||||||
r = requests.get(f"{BASE_URL}/avatar/multiavatar/clamp_low?size=1")
|
|
||||||
assert r.status_code == 200
|
|
||||||
assert 'width="16"' in r.text
|
|
||||||
|
|
||||||
|
|
||||||
def test_avatar_size_clamped_high(app_server):
|
|
||||||
r = requests.get(f"{BASE_URL}/avatar/multiavatar/clamp_high?size=9999")
|
|
||||||
assert r.status_code == 200
|
|
||||||
assert 'width="512"' in r.text
|
|
||||||
|
|||||||
@@ -89,6 +89,53 @@ def test_run_clamps_max_pages_above_cap(app_server):
|
|||||||
_clear_seo_jobs()
|
_clear_seo_jobs()
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_rejects_malformed_scheme(app_server):
|
||||||
|
r = requests.post(
|
||||||
|
f"{BASE_URL}/tools/seo/run",
|
||||||
|
headers=_json_headers(),
|
||||||
|
data={"url": "ahttps://devplace.net/sitem", "mode": "url"},
|
||||||
|
allow_redirects=False,
|
||||||
|
)
|
||||||
|
assert r.status_code in (400, 422), r.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_rejects_relative_path(app_server):
|
||||||
|
r = requests.post(
|
||||||
|
f"{BASE_URL}/tools/seo/run",
|
||||||
|
headers=_json_headers(),
|
||||||
|
data={"url": "/feed", "mode": "url"},
|
||||||
|
allow_redirects=False,
|
||||||
|
)
|
||||||
|
assert r.status_code in (400, 422), r.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_normalizes_missing_scheme(app_server):
|
||||||
|
try:
|
||||||
|
r = requests.post(
|
||||||
|
f"{BASE_URL}/tools/seo/run",
|
||||||
|
headers=_json_headers(),
|
||||||
|
data={"url": "example.com", "mode": "url", "max_pages": "5"},
|
||||||
|
)
|
||||||
|
assert r.status_code == 200, r.text
|
||||||
|
uid = r.json()["uid"]
|
||||||
|
refresh_snapshot()
|
||||||
|
job = queue.get_job(uid)
|
||||||
|
assert job is not None
|
||||||
|
assert job["payload"]["url"] == "https://example.com"
|
||||||
|
finally:
|
||||||
|
_clear_seo_jobs()
|
||||||
|
|
||||||
|
|
||||||
|
def test_run_rejects_non_http_scheme(app_server):
|
||||||
|
r = requests.post(
|
||||||
|
f"{BASE_URL}/tools/seo/run",
|
||||||
|
headers=_json_headers(),
|
||||||
|
data={"url": "ftp://example.com", "mode": "url"},
|
||||||
|
allow_redirects=False,
|
||||||
|
)
|
||||||
|
assert r.status_code in (400, 422), r.text
|
||||||
|
|
||||||
|
|
||||||
def test_run_enforces_one_active_job_per_owner(app_server):
|
def test_run_enforces_one_active_job_per_owner(app_server):
|
||||||
try:
|
try:
|
||||||
first = requests.post(
|
first = requests.post(
|
||||||
|
|||||||
Reference in New Issue
Block a user