102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
"""Load the Catbox userhash and upload one file.
|
|
|
|
The hash is read from ``secrets/credentials.toml``. It never appears on a
|
|
process command line or in printed output.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import ssl
|
|
import tomllib
|
|
from pathlib import Path
|
|
from secrets import token_hex
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
API: str = "https://catbox.moe/user/api.php"
|
|
TIMEOUT_SECS: int = 180
|
|
|
|
|
|
def load_userhash(vault: Path) -> str:
|
|
"""Return the Catbox ``userhash`` from ``vault/secrets/credentials.toml``."""
|
|
path = vault / "secrets" / "credentials.toml"
|
|
if not path.is_file():
|
|
raise RuntimeError(f"missing {path}")
|
|
text = path.read_text(encoding="utf-8")
|
|
try:
|
|
data = tomllib.loads(text)
|
|
value = str((data.get("catbox") or {}).get("userhash") or "").strip()
|
|
except tomllib.TOMLDecodeError:
|
|
value = ""
|
|
in_catbox = False
|
|
for raw in text.splitlines():
|
|
line = raw.strip()
|
|
if line.startswith("[") and line.endswith("]"):
|
|
in_catbox = line[1:-1].strip().lower() == "catbox"
|
|
continue
|
|
if in_catbox and line.lower().startswith("userhash"):
|
|
_key, _equals, remainder = line.partition("=")
|
|
value = remainder.strip().strip("\"'")
|
|
break
|
|
if not value:
|
|
raise RuntimeError("credentials.toml [catbox] userhash is empty")
|
|
return value
|
|
|
|
|
|
def _encode_multipart(
|
|
fields: list[tuple[str, str]],
|
|
file_field: str,
|
|
file_path: Path,
|
|
) -> tuple[bytes, str]:
|
|
"""Build a multipart body. Field values stay in the body, not in argv."""
|
|
boundary = "----Threadwell" + token_hex(16)
|
|
chunks: list[bytes] = []
|
|
for name, value in fields:
|
|
chunks.append(
|
|
(
|
|
f"--{boundary}\r\n"
|
|
f'Content-Disposition: form-data; name="{name}"\r\n'
|
|
f"\r\n"
|
|
f"{value}\r\n"
|
|
).encode("utf-8")
|
|
)
|
|
filename = file_path.name.replace('"', "_")
|
|
chunks.append(
|
|
(
|
|
f"--{boundary}\r\n"
|
|
f'Content-Disposition: form-data; name="{file_field}"; '
|
|
f'filename="{filename}"\r\n'
|
|
f"Content-Type: application/octet-stream\r\n"
|
|
f"\r\n"
|
|
).encode("utf-8")
|
|
)
|
|
chunks.append(file_path.read_bytes())
|
|
chunks.append(f"\r\n--{boundary}--\r\n".encode("utf-8"))
|
|
return b"".join(chunks), f"multipart/form-data; boundary={boundary}"
|
|
|
|
|
|
def upload_file(path: Path, userhash: str) -> str:
|
|
"""POST ``path`` to catbox.moe and return the public HTTPS URL."""
|
|
body, content_type = _encode_multipart(
|
|
[("reqtype", "fileupload"), ("userhash", userhash)],
|
|
"fileToUpload",
|
|
path,
|
|
)
|
|
request = Request(API, data=body, method="POST")
|
|
request.add_header("Content-Type", content_type)
|
|
try:
|
|
with urlopen(
|
|
request,
|
|
timeout=TIMEOUT_SECS,
|
|
context=ssl.create_default_context(),
|
|
) as resp:
|
|
url = resp.read().decode("utf-8", errors="replace").strip()
|
|
except HTTPError as exc:
|
|
raise RuntimeError(
|
|
f"catbox upload failed for {path.name} (HTTP {exc.code})"
|
|
) from None
|
|
except URLError:
|
|
raise RuntimeError(f"catbox upload failed for {path.name}") from None
|
|
if not url.startswith("https://"):
|
|
raise RuntimeError(f"catbox refused {path.name}: non-URL response")
|
|
return url
|