From 06186b444a7a8d9d3c78a290c3b3584b47c5e4df Mon Sep 17 00:00:00 2001 From: Nikolay Bryskin Date: Wed, 22 Jul 2026 02:49:57 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20add=20`rdt=20post`=20=E2=80=94=20create?= =?UTF-8?q?=20post=20drafts=20and=20(captcha-gated)=20publish?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a `rdt post` command that creates Reddit posts through Reddit's web GraphQL endpoint (svc/shreddit/graphql), reusing the existing cookie session. Drafts (text/link) are created fully headlessly. Publishing (subreddit `CreatePost` / profile `CreateProfilePost`, incl. images) is gated behind reCAPTCHA Enterprise, so it requires a browser-supplied `--recaptcha-token`; the tool never solves the captcha itself. Image drafts are rejected because Reddit's draft input type has no media field. Key pieces: - transports: TLS fix β€” httpx's default cipher list produces a ClientHello that Reddit's WAF blocks on the GraphQL endpoint (403 "blocked by network security"); reset to ssl.create_default_context() + set_ciphers("DEFAULT"). Pure-Python, no new deps, not impersonation. Also allow application/json writes on the write transport. - session: ensure_csrf_token() β€” double-submit CSRF (same value in cookie and JSON body; generated if absent). - client: _graphql, resolve_subreddit_id, create_media_lease, upload_image (S3 multipart), create_post, create_draft. - commands/submit.py: the `post` command (--text/--url/--image, --draft, --recaptcha-token, --nsfw/--spoiler, --json/--yaml; profile via u_). - tests + README/SKILL docs. No new runtime dependencies. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 17 ++- SKILL.md | 17 ++- rdt_cli/cli.py | 6 +- rdt_cli/client.py | 186 +++++++++++++++++++++++++++- rdt_cli/commands/submit.py | 173 ++++++++++++++++++++++++++ rdt_cli/constants.py | 24 ++++ rdt_cli/session.py | 17 +++ rdt_cli/transports.py | 34 +++++- tests/test_cli.py | 243 ++++++++++++++++++++++++++++++++++++- 9 files changed, 708 insertions(+), 9 deletions(-) create mode 100644 rdt_cli/commands/submit.py diff --git a/README.md b/README.md index 4373ace..21cd9fb 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ A CLI for Reddit β€” browse feeds, read posts, search, and interact via reverse- - πŸ“€ **Export** β€” export search results to CSV or JSON; `-o file.json` on any listing - πŸ‘€ **Users** β€” view user profiles, post history, comment history, saved and upvoted items - ⬆️ **Interactions** β€” upvote/downvote, save/unsave, subscribe/unsubscribe, comment (with 1.5-4s rate-limit delay) +- ✍️ **Create posts** β€” save text/link **drafts** headlessly (`rdt post --draft`); publishing is reCAPTCHA-gated and needs a browser token (`--recaptcha-token`) - πŸ›‘οΈ **Anti-detection** β€” consistent Chrome 133 fingerprint, `sec-ch-ua` alignment, Gaussian jitter, exponential backoff - πŸ“Š **Structured output** β€” `--yaml`, `--json`, `--output FILE`, `--compact`, `--full-text` - πŸ“¦ **Stable envelope** β€” see [SCHEMA.md](./SCHEMA.md) for `ok/schema_version/data/error` @@ -120,8 +121,21 @@ rdt save 3 --undo # Unsave rdt subscribe python # Subscribe to r/python rdt subscribe python --undo # Unsubscribe rdt comment 3 "Great post!" # Comment on result #3 + +# ─── Create posts (require login) ───────────────── +# Drafts save headlessly. Publishing is gated behind reCAPTCHA Enterprise, so it +# needs a --recaptcha-token captured from a browser (single-use, ~2 min). +rdt post python "WIP title" --text "Hello **world**" --draft # Save a text draft +rdt post news "Cool read" --url https://example.com --draft # Save a link draft +rdt post python "Title" --text "body" --recaptcha-token # Publish text post +rdt post pics "My cat" --image cat.jpg --recaptcha-token # Publish image post +rdt post u_yourname "On my profile" --text "hi" --recaptcha-token ``` +> **Note:** Reddit drafts store text/link only β€” not images (image drafts aren't +> supported by Reddit). And publishing any post requires a reCAPTCHA Enterprise +> token that only a browser can produce; rdt-cli never solves the captcha. + ## Authentication rdt-cli supports browser cookie extraction to authenticate with Reddit: @@ -219,7 +233,8 @@ rdt_cli/ β”œβ”€β”€ browse.py # feed, popular, all, sub, sub-info, user, user-posts, user-comments, saved, upvoted, open β”œβ”€β”€ post.py # read, show β”œβ”€β”€ search.py # search, export - └── social.py # upvote, save, subscribe, comment + β”œβ”€β”€ social.py # upvote, save, subscribe, comment + └── submit.py # post (text/link/image, drafts) ``` ## Development diff --git a/SKILL.md b/SKILL.md index 512e2c3..4ef195a 100644 --- a/SKILL.md +++ b/SKILL.md @@ -73,7 +73,7 @@ Payloads live under `.data`. - `--output file.json` β†’ save structured output to file - Rich output β†’ **stderr** (safe for pipes: `rdt search X --json | jq .data`) - Most read commands work without auth (public Reddit JSON API) -- Write actions (upvote, save, subscribe) require auth + built-in 1.5-4s delay +- Write actions (upvote, save, subscribe, comment, post) require auth + built-in 1.5-4s delay ## Command Reference @@ -126,6 +126,21 @@ Payloads live under `.data`. | `rdt subscribe --undo` | Unsubscribe | `rdt subscribe python --undo` | | `rdt comment ` | Post a comment | `rdt comment 3 "Great post!"` | +### Create posts (require auth) + +Exactly one of `--text` / `--url` / `--image`. Target a subreddit or a profile (`u_`). Add `--json` for a structured result. + +**Drafts** save headlessly. **Publishing** is gated behind reCAPTCHA Enterprise (invisible, score-based) β€” it needs a `--recaptcha-token` captured from a browser (single-use, ~2 min). rdt-cli never solves the captcha. + +| Command | Description | Example | +|---------|-------------|---------| +| `rdt post --text <body> --draft` | Save a text draft | `rdt post python "WIP" --text "..." --draft` | +| `rdt post <sub> <title> --url <url> --draft` | Save a link draft | `rdt post news "Read" --url https://ex.com --draft` | +| `rdt post <sub> <title> --text <body> --recaptcha-token <tok>` | Publish text post | `rdt post python "Hi" --text "..." --recaptcha-token <tok>` | +| `rdt post <sub> <title> --image <file> --recaptcha-token <tok>` | Publish image post | `rdt post pics "Cat" --image cat.jpg --recaptcha-token <tok>` | + +Optional flags: `--nsfw`, `--spoiler`. Notes: Reddit drafts can't store images (text/link only); `--image` requires publishing with a token. + ### Account | Command | Description | diff --git a/rdt_cli/cli.py b/rdt_cli/cli.py index 2a3cf55..4ccb3eb 100644 --- a/rdt_cli/cli.py +++ b/rdt_cli/cli.py @@ -17,7 +17,7 @@ import click from . import __version__ -from .commands import auth, browse, post, search, social +from .commands import auth, browse, post, search, social, submit @click.group() @@ -71,6 +71,10 @@ def cli(ctx: click.Context, verbose: bool) -> None: cli.add_command(social.subscribe) cli.add_command(social.comment) +# ─── Create commands ──────────────────────────────────────────────── + +cli.add_command(submit.post) + if __name__ == "__main__": cli() diff --git a/rdt_cli/client.py b/rdt_cli/client.py index 7762976..e84f981 100644 --- a/rdt_cli/client.py +++ b/rdt_cli/client.py @@ -3,15 +3,28 @@ from __future__ import annotations import logging +import mimetypes +import uuid +from pathlib import Path from typing import Any +import httpx + from .config import DEFAULT_CONFIG, RuntimeConfig from .constants import ( ALL_URL, + BASE_URL, COMMENT_URL, DEFAULT_LIMIT, + GRAPHQL_URL, HOME_URL, + IMAGE_MIME_TYPES, + MEDIA_S3_HOST, MORECHILDREN_URL, + OP_CREATE_DRAFT, + OP_CREATE_POST, + OP_CREATE_PROFILE_POST, + OP_MEDIA_LEASE, POPULAR_URL, POST_COMMENTS_SHORT_URL, POST_COMMENTS_URL, @@ -34,7 +47,7 @@ ) from .fingerprint import BrowserFingerprint from .session import SessionState -from .transports import ReadTransport, WriteTransport +from .transports import SSL_CONTEXT, ReadTransport, WriteTransport logger = logging.getLogger(__name__) @@ -351,6 +364,177 @@ def post_comment(self, parent_fullname: str, text: str) -> dict: """Post a comment.""" return self._post(COMMENT_URL, data={"parent": parent_fullname, "text": text}) + # ── Post creation (GraphQL) ───────────────────────────────────── + + def _graphql(self, operation: str, variables: dict[str, Any]) -> Any: + """POST a shreddit GraphQL operation and return its ``data`` payload. + + Sends ``{operation, variables, csrf_token}`` as JSON. The csrf_token is + echoed into the write transport's cookie jar so cookie == body (Reddit's + double-submit CSRF). Raises RedditApiError on GraphQL-level errors. + """ + if self._write_transport is None: + raise RuntimeError("Client not initialized. Use 'with RedditClient() as client:'") + token = self.session.ensure_csrf_token() + self._write_transport.client.cookies.set("csrf_token", token) + payload = {"operation": operation, "variables": variables, "csrf_token": token} + result = self._write_request("POST", GRAPHQL_URL, json=payload) + if isinstance(result, dict) and result.get("errors"): + errors = result["errors"] + msg = "; ".join( + e.get("message", str(e)) if isinstance(e, dict) else str(e) for e in errors + ) or str(errors) + raise RedditApiError(f"GraphQL {operation} failed: {msg}") + if isinstance(result, dict): + return result.get("data", result) + return result + + @staticmethod + def _markdown_content(body: str | None) -> dict[str, Any]: + """Content payload for a markdown body (empty dict for no body).""" + return {"markdown": body} if body else {} + + def resolve_subreddit_id(self, subreddit: str) -> str: + """Resolve a subreddit name to its ``t5_`` fullname (for GraphQL input).""" + about = self.get_subreddit_about(subreddit) + sub_id = about.get("name") if isinstance(about, dict) else None + if not sub_id: + raise RedditApiError(f"Could not resolve subreddit r/{subreddit}") + return sub_id + + def create_media_lease(self, mimetype_token: str) -> dict[str, Any]: + """Request an S3 upload lease for an image (``mimetype_token`` e.g. 'JPEG').""" + data = self._graphql(OP_MEDIA_LEASE, {"input": {"mimetype": mimetype_token}}) + lease = data.get("createMediaUploadLease") if isinstance(data, dict) else None + if not isinstance(lease, dict) or not lease.get("uploadLease"): + raise RedditApiError("Media lease request returned no upload lease") + return lease + + def upload_image(self, path: str) -> str: + """Upload an image to Reddit's S3 and return its media asset id. + + Flow: lease (GraphQL) β†’ multipart POST to the S3 URL from the lease + (lease fields verbatim + ``file`` last, no reddit cookies) β†’ mediaId. + """ + file_path = Path(path) + if not file_path.is_file(): + raise RedditApiError(f"Image file not found: {path}") + content_type, _ = mimetypes.guess_type(file_path.name) + if content_type not in IMAGE_MIME_TYPES: + raise RedditApiError( + f"Unsupported image type for {path} (allowed: {', '.join(sorted(IMAGE_MIME_TYPES))})" + ) + token = content_type.split("/")[-1].upper() # image/jpeg β†’ JPEG + + lease = self.create_media_lease(token) + upload_lease = lease["uploadLease"] + upload_url = upload_lease["uploadLeaseUrl"] + if upload_url.startswith("//"): + upload_url = "https:" + upload_url + fields = {h["header"]: h["value"] for h in upload_lease.get("uploadLeaseHeaders", [])} + media_id = lease.get("mediaId") + if not media_id: + raise RedditApiError("Media lease did not return a mediaId") + + # Separate client: cross-site S3 host, must NOT carry reddit cookies. + headers = { + "User-Agent": self._fingerprint.user_agent, + "Origin": BASE_URL, + "Referer": f"{BASE_URL}/", + } + with httpx.Client( + follow_redirects=True, timeout=httpx.Timeout(self._timeout), verify=SSL_CONTEXT + ) as s3: + resp = s3.post( + upload_url, + data=fields, + files={"file": (file_path.name, file_path.read_bytes(), content_type)}, + headers=headers, + ) + if resp.status_code not in (200, 201): + raise RedditApiError(f"Image upload failed (HTTP {resp.status_code})") + return media_id + + def create_post( + self, + subreddit_id: str, + title: str, + *, + recaptcha_token: str, + kind: str = "self", + body: str | None = None, + url: str | None = None, + media_id: str | None = None, + nsfw: bool = False, + spoiler: bool = False, + is_profile: bool = False, + ) -> Any: + """Publish a post (subreddit ``CreatePost`` / profile ``CreateProfilePost``). + + Reddit gates post submission behind **reCAPTCHA Enterprise** (invisible, + score-based, action ``post_submit``). ``recaptcha_token`` MUST be a fresh + token obtained from a browser (single-use, ~2 min TTL) β€” it cannot be + produced headlessly, so this call fails without one. Field shapes come + from captured live requests + ``ValidateCreatePostInput``: + selfβ†’``content.markdown``, linkβ†’``url``, imageβ†’``gallery.items[].mediaId`` + (subreddit) or ``image.url`` (profile). + """ + inp: dict[str, Any] = { + "title": title, + "isNsfw": bool(nsfw), + "isSpoiler": bool(spoiler), + "content": self._markdown_content(body) if kind == "self" else {}, + "recaptchaToken": recaptcha_token, + "correlationId": str(uuid.uuid4()), + } + if is_profile: + inp["isCommercialCommunication"] = False + inp["targetLanguage"] = "" + if kind == "link": + inp["url"] = url + elif kind == "image": + inp["image"] = {"url": f"{MEDIA_S3_HOST}/{media_id}"} + return self._graphql(OP_CREATE_PROFILE_POST, {"input": inp}) + + inp["subredditId"] = subreddit_id + inp["postType"] = {"self": "TEXT", "link": "LINK", "image": "IMAGE"}[kind] + if kind == "link": + inp["url"] = url + elif kind == "image": + inp["gallery"] = {"items": [{"mediaId": media_id}]} + return self._graphql(OP_CREATE_POST, {"input": inp}) + + def create_draft( + self, + subreddit_id: str, + title: str, + *, + body: str | None = None, + url: str | None = None, + nsfw: bool = False, + spoiler: bool = False, + ) -> Any: + """Save a post draft via ``CreateDraft`` (text or link β€” no captcha needed). + + Reddit drafts store text (markdown) or a link URL only; they cannot hold + images (the draft input type has no media field). Text is confirmed live; + the link shape is best-effort. + """ + inp: dict[str, Any] = { + "subredditId": subreddit_id, + "title": title, + "isNsfw": bool(nsfw), + "isSpoiler": bool(spoiler), + } + if url: + inp["kind"] = "LINK" + inp["url"] = url + inp["content"] = {} + else: + inp["kind"] = "MARKDOWN" + inp["content"] = self._markdown_content(body) + return self._graphql(OP_CREATE_DRAFT, {"input": inp}) + # ── Subscription feed ─────────────────────────────────────────── def get_my_subscriptions( diff --git a/rdt_cli/commands/submit.py b/rdt_cli/commands/submit.py new file mode 100644 index 0000000..2b7ae7b --- /dev/null +++ b/rdt_cli/commands/submit.py @@ -0,0 +1,173 @@ +"""Post creation: draft (headless) and publish (reCAPTCHA-gated) via GraphQL. + +Reddit's web GraphQL (see rdt_cli.client) lets us create **drafts** headlessly, +but **publishing** a post is gated behind reCAPTCHA Enterprise (invisible, +score-based, action ``post_submit``). A valid token can only be produced by a +real browser, so publishing requires the caller to pass ``--recaptcha-token`` +(obtained from DevTools, single-use, ~2 min TTL). The tool never solves the +captcha itself. Without a token, use ``--draft``. +""" + +from __future__ import annotations + +from typing import Any + +import click + +from ..client import RedditClient +from ..constants import BASE_URL +from ..exceptions import RedditApiError +from ._common import ( + console, + exit_for_error, + maybe_print_structured, + require_auth, + structured_output_options, + write_delay, +) + + +def _find_first(obj: Any, keys: set[str]) -> str | None: + """Depth-first search for the first non-empty string under any of ``keys``.""" + if isinstance(obj, dict): + for key, value in obj.items(): + if key in keys and isinstance(value, str) and value: + return value + for value in obj.values(): + found = _find_first(value, keys) + if found: + return found + elif isinstance(obj, list): + for item in obj: + found = _find_first(item, keys) + if found: + return found + return None + + +def _permalink_from(data: Any) -> str | None: + """Best-effort extraction of a post permalink from a GraphQL response.""" + permalink = _find_first(data, {"permalink"}) or _find_first(data, {"postUrl"}) + if permalink and permalink.startswith("/"): + permalink = f"{BASE_URL}{permalink}" + return permalink + + +_RECAPTCHA_HELP = ( + "Publishing requires a reCAPTCHA Enterprise token β€” Reddit gates post " + "submission with invisible, score-based reCAPTCHA (action 'post_submit'). " + "A token can only be produced by a browser (single-use, ~2 min). Pass it via " + "--recaptcha-token, or use --draft to save a draft headlessly." +) + + +@click.command() +@click.argument("subreddit") +@click.argument("title") +@click.option("--text", "text", default=None, help="Self/text post body (markdown)") +@click.option("--url", "link_url", default=None, help="Link post URL") +@click.option( + "--image", "image", + type=click.Path(exists=True, dir_okay=False), + default=None, + help="Image file to post (publish only; drafts can't store images)", +) +@click.option("--draft", is_flag=True, help="Save as a draft instead of publishing (text/link only)") +@click.option( + "--recaptcha-token", "recaptcha_token", default=None, + help="reCAPTCHA Enterprise token from a browser (required to publish)", +) +@click.option("--nsfw", is_flag=True, help="Mark as NSFW") +@click.option("--spoiler", is_flag=True, help="Mark as spoiler") +@structured_output_options +def post( + subreddit: str, + title: str, + text: str | None, + link_url: str | None, + image: str | None, + draft: bool, + recaptcha_token: str | None, + nsfw: bool, + spoiler: bool, + as_json: bool, + as_yaml: bool, +) -> None: + """Create a post in a subreddit or profile (u_<name>): text, link, or image. + + Provide exactly one of --text, --url, or --image. + + Drafts (--draft) are created headlessly. Publishing is gated behind reCAPTCHA + Enterprise, so it needs a --recaptcha-token captured from a browser. + + Examples: + rdt post python "My title" --text "Hello **world**" --draft + rdt post news "Interesting" --url https://example.com --draft + rdt post pics "My cat" --image cat.jpg --recaptcha-token <token> + rdt post u_myname "On my profile" --text "hi" --recaptcha-token <token> + """ + provided = [ + name for name, given in + (("--text", text is not None), ("--url", link_url is not None), ("--image", image is not None)) + if given + ] + if len(provided) != 1: + raise click.UsageError("Provide exactly one of --text, --url, or --image.") + kind = {"--text": "self", "--url": "link", "--image": "image"}[provided[0]] + + if draft and kind == "image": + raise click.UsageError( + "--image cannot be combined with --draft: Reddit drafts cannot store " + "images (the image only attaches at publish time). Publish it with " + "--recaptcha-token, or draft text/link only." + ) + if not draft and not recaptcha_token: + raise click.UsageError(_RECAPTCHA_HELP) + + is_profile = subreddit.lower().startswith("u_") + + cred = require_auth() + try: + with RedditClient(cred) as client: + client.validate_session() + subreddit_id = client.resolve_subreddit_id(subreddit) + if draft: + data = client.create_draft( + subreddit_id, title, body=text, url=link_url, nsfw=nsfw, spoiler=spoiler, + ) + action = "Draft saved" + else: + media_id = client.upload_image(image) if image is not None else None + data = client.create_post( + subreddit_id, title, recaptcha_token=recaptcha_token or "", kind=kind, + body=text, url=link_url, media_id=media_id, nsfw=nsfw, spoiler=spoiler, + is_profile=is_profile, + ) + action = "Posted" + write_delay() + + permalink = _permalink_from(data) + draft_id = _find_first(data, {"id"}) if draft else None + payload: dict[str, Any] = { + "action": "draft" if draft else "post", + "subreddit": subreddit, + "title": title, + "kind": kind, + "result": data, + } + if permalink: + payload["permalink"] = permalink + if draft_id: + payload["draft_id"] = draft_id + + if maybe_print_structured(payload, as_json=as_json, as_yaml=as_yaml): + return + target = permalink or draft_id or f'"{title}"' + console.print(f"[green]βœ… {action}[/green] to {subreddit}: {target}") + if draft: + console.print( + f"[dim]Drafts can't be published headlessly (reCAPTCHA). " + f"Finish in browser: {BASE_URL}/r/{subreddit}/submit[/dim]" + ) + except RedditApiError as exc: + exit_for_error(exc, as_json=as_json, as_yaml=as_yaml, prefix="Post failed") diff --git a/rdt_cli/constants.py b/rdt_cli/constants.py index 055064b..0da40e1 100644 --- a/rdt_cli/constants.py +++ b/rdt_cli/constants.py @@ -51,6 +51,30 @@ COMMENT_URL = "/api/comment" SUBSCRIPTIONS_URL = "/subreddits/mine/subscriber.json" +# ── Post creation (Reddit web GraphQL) ────────────────────────────── +# Modern Reddit ("shreddit") creates posts/drafts and leases media uploads +# through a single GraphQL endpoint on www.reddit.com. JSON body is +# {operation, variables, csrf_token}; csrf_token is double-submit (same value +# in the cookie jar and the body). See rdt_cli.client for the flow. +GRAPHQL_URL = "/svc/shreddit/graphql" +OP_CREATE_POST = "CreatePost" # subreddit posts +OP_CREATE_PROFILE_POST = "CreateProfilePost" # posts to u_<username> profile +OP_CREATE_DRAFT = "CreateDraft" +OP_MEDIA_LEASE = "CreateMediaUploadLease" + +# S3 host that serves uploaded media; the object URL is "<host>/<mediaId>". +MEDIA_S3_HOST = "https://reddit-uploaded-media.s3-accelerate.amazonaws.com" + +# Image posts: allowed content types (extension β†’ MIME). The lease token sent +# to Reddit is the uppercased subtype (e.g. image/jpeg β†’ "JPEG"); the actual S3 +# upload URL and fields come from the lease response. +IMAGE_MIME_TYPES = { + "image/png", + "image/jpeg", + "image/gif", + "image/webp", +} + # ── Request Headers (Chrome 133, macOS) ───────────────────────────── HEADERS = { "User-Agent": ( diff --git a/rdt_cli/session.py b/rdt_cli/session.py index 8cda2c9..a1e8fe2 100644 --- a/rdt_cli/session.py +++ b/rdt_cli/session.py @@ -2,6 +2,7 @@ from __future__ import annotations +import secrets from dataclasses import dataclass, field from typing import Any @@ -63,6 +64,22 @@ def refresh_capabilities(self) -> None: self.capabilities = capabilities + def ensure_csrf_token(self) -> str: + """Return a csrf_token for GraphQL writes, generating one if absent. + + Reddit's web GraphQL uses a double-submit CSRF token: the same value + must appear in both the request cookie and the JSON body, and it may be + client-generated (only ``reddit_session`` must be a real login cookie). + Prefer an existing ``csrf_token`` cookie so we match the browser; fall + back to a fresh token otherwise. + """ + token = self.cookies.get("csrf_token") + if not token: + token = secrets.token_hex(16) + self.cookies["csrf_token"] = token + self.refresh_capabilities() + return token + def apply_identity(self, identity: dict[str, Any]) -> None: """Update session from a validated identity payload.""" data = identity.get("data", identity) diff --git a/rdt_cli/transports.py b/rdt_cli/transports.py index 8f135b2..76a3562 100644 --- a/rdt_cli/transports.py +++ b/rdt_cli/transports.py @@ -4,6 +4,7 @@ import logging import random +import ssl import time from typing import Any @@ -24,6 +25,24 @@ logger = logging.getLogger(__name__) +def build_ssl_context() -> ssl.SSLContext: + """SSL context whose TLS fingerprint Reddit's WAF accepts. + + httpx/httpcore ship a curated, restricted cipher list. The resulting + ClientHello fingerprint is rejected (HTTP 403 "blocked by network security") + by Reddit's edge on POST /svc/shreddit/graphql β€” while reads and the legacy + /api/* endpoints are unaffected. Python's stock cipher set passes, so we + build a default context and reset it to OpenSSL's DEFAULT ciphers. Cert + verification is preserved. + """ + context = ssl.create_default_context() + context.set_ciphers("DEFAULT") + return context + + +SSL_CONTEXT = build_ssl_context() + + class BaseTransport: """Shared retry, throttling, and cookie management.""" @@ -48,6 +67,7 @@ def __init__( cookies=session.cookies, follow_redirects=True, timeout=httpx.Timeout(config.timeout), + verify=SSL_CONTEXT, ) def close(self) -> None: @@ -161,10 +181,16 @@ def request(self, method: str, url: str, **kwargs: Any) -> Any: headers = dict(kwargs.pop("headers", {})) headers.update(self.fingerprint.write_headers(modhash=self.session.modhash)) - kwargs["headers"] = headers - data = kwargs.get("data") - if isinstance(data, dict) and self.session.modhash and "uh" not in data: - kwargs["data"] = {**data, "uh": self.session.modhash} + if "json" in kwargs: + # GraphQL / JSON writes (svc/shreddit/graphql): send application/json + # and do not inject the form-style `uh` modhash. CSRF is carried in + # the JSON body + cookie by the caller. + headers["Content-Type"] = "application/json" + else: + data = kwargs.get("data") + if isinstance(data, dict) and self.session.modhash and "uh" not in data: + kwargs["data"] = {**data, "uh": self.session.modhash} + kwargs["headers"] = headers return super().request(method, url, **kwargs) diff --git a/tests/test_cli.py b/tests/test_cli.py index 50e7df2..d96c3bd 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -37,6 +37,7 @@ def test_all_commands_registered(self): "read", "show", "search", "export", "upvote", "save", "subscribe", "comment", + "post", ] for cmd in expected: assert cmd in result.output, f"Missing command: {cmd}" @@ -52,7 +53,7 @@ def test_command_count(self): # Count command lines (indented, after "Commands:" ) lines = result.output.split("\n") cmd_lines = [line for line in lines if line.startswith(" ") and not line.strip().startswith("-")] - assert len(cmd_lines) >= 22 + assert len(cmd_lines) >= 23 # ── Command help ──────────────────────────────────────────────────── @@ -70,6 +71,7 @@ class TestCommandHelp: "read", "show", "search", "export", "upvote", "save", "subscribe", "comment", + "post", ], ) def test_help(self, cmd): @@ -972,3 +974,242 @@ def test_show_help_shows_compact(self): assert result.exit_code == 0 assert "--compact" in result.output + +# ── Post creation command (mocked) ────────────────────────────────── + + +class TestPostCommand: + """Test the `post` creation command with mocked client methods.""" + + def _cred(self): + from rdt_cli.auth import Credential + + return Credential(cookies={"reddit_session": "x", "csrf_token": "y"}, username="me") + + def test_text_draft(self): + cred = self._cred() + with patch("rdt_cli.commands._common.get_credential", return_value=cred), \ + patch("rdt_cli.commands.submit.write_delay"), \ + patch("rdt_cli.client.RedditClient.validate_session", return_value={}), \ + patch("rdt_cli.client.RedditClient.resolve_subreddit_id", return_value="t5_abc"), \ + patch( + "rdt_cli.client.RedditClient.create_draft", + return_value={"createPostDraft": {"ok": True, "postDraft": {"id": "d-123"}}}, + ) as mock_draft, \ + patch("rdt_cli.client.RedditClient.create_post") as mock_post: + result = runner.invoke(cli, ["post", "test", "WIP", "--text", "body", "--draft", "--json"]) + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["data"]["action"] == "draft" + assert data["data"]["draft_id"] == "d-123" + _, kwargs = mock_draft.call_args + assert kwargs["body"] == "body" + mock_post.assert_not_called() + + def test_link_draft(self): + cred = self._cred() + with patch("rdt_cli.commands._common.get_credential", return_value=cred), \ + patch("rdt_cli.commands.submit.write_delay"), \ + patch("rdt_cli.client.RedditClient.validate_session", return_value={}), \ + patch("rdt_cli.client.RedditClient.resolve_subreddit_id", return_value="t5_abc"), \ + patch("rdt_cli.client.RedditClient.create_draft", return_value={}) as mock_draft: + result = runner.invoke( + cli, ["post", "test", "A link", "--url", "https://example.com", "--draft", "--json"] + ) + assert result.exit_code == 0, result.output + _, kwargs = mock_draft.call_args + assert kwargs["url"] == "https://example.com" + + def test_publish_requires_recaptcha_token(self): + cred = self._cred() + with patch("rdt_cli.commands._common.get_credential", return_value=cred): + result = runner.invoke(cli, ["post", "test", "T", "--text", "hi"]) + assert result.exit_code == 2 + assert "reCAPTCHA" in result.output + + def test_text_publish_with_token(self): + cred = self._cred() + with patch("rdt_cli.commands._common.get_credential", return_value=cred), \ + patch("rdt_cli.commands.submit.write_delay"), \ + patch("rdt_cli.client.RedditClient.validate_session", return_value={}), \ + patch("rdt_cli.client.RedditClient.resolve_subreddit_id", return_value="t5_abc"), \ + patch("rdt_cli.client.RedditClient.create_post", return_value={}) as mock_post: + result = runner.invoke( + cli, ["post", "test", "T", "--text", "hi", "--recaptcha-token", "TOK", "--json"] + ) + assert result.exit_code == 0, result.output + _, kwargs = mock_post.call_args + assert kwargs["kind"] == "self" + assert kwargs["recaptcha_token"] == "TOK" + assert kwargs["is_profile"] is False + + def test_image_publish_with_token(self, tmp_path): + cred = self._cred() + img = tmp_path / "pic.png" + img.write_bytes(b"\x89PNG\r\n\x1a\n") + with patch("rdt_cli.commands._common.get_credential", return_value=cred), \ + patch("rdt_cli.commands.submit.write_delay"), \ + patch("rdt_cli.client.RedditClient.validate_session", return_value={}), \ + patch("rdt_cli.client.RedditClient.resolve_subreddit_id", return_value="t5_abc"), \ + patch("rdt_cli.client.RedditClient.upload_image", return_value="media123") as mock_up, \ + patch("rdt_cli.client.RedditClient.create_post", return_value={}) as mock_post: + result = runner.invoke( + cli, ["post", "test", "Pic", "--image", str(img), "--recaptcha-token", "TOK", "--json"] + ) + assert result.exit_code == 0, result.output + mock_up.assert_called_once_with(str(img)) + _, kwargs = mock_post.call_args + assert kwargs["kind"] == "image" + assert kwargs["media_id"] == "media123" + + def test_profile_publish_uses_profile_flag(self): + cred = self._cred() + with patch("rdt_cli.commands._common.get_credential", return_value=cred), \ + patch("rdt_cli.commands.submit.write_delay"), \ + patch("rdt_cli.client.RedditClient.validate_session", return_value={}), \ + patch("rdt_cli.client.RedditClient.resolve_subreddit_id", return_value="t5_prof"), \ + patch("rdt_cli.client.RedditClient.create_post", return_value={}) as mock_post: + result = runner.invoke( + cli, ["post", "u_me", "T", "--text", "hi", "--recaptcha-token", "TOK", "--json"] + ) + assert result.exit_code == 0, result.output + _, kwargs = mock_post.call_args + assert kwargs["is_profile"] is True + + def test_requires_a_kind(self): + cred = self._cred() + with patch("rdt_cli.commands._common.get_credential", return_value=cred): + result = runner.invoke(cli, ["post", "test", "No body", "--draft"]) + assert result.exit_code == 2 + assert "exactly one" in result.output + + def test_rejects_two_kinds(self): + cred = self._cred() + with patch("rdt_cli.commands._common.get_credential", return_value=cred): + result = runner.invoke( + cli, ["post", "test", "Two", "--text", "a", "--url", "https://e.com", "--draft"] + ) + assert result.exit_code == 2 + assert "exactly one" in result.output + + def test_rejects_image_draft(self, tmp_path): + cred = self._cred() + img = tmp_path / "pic.png" + img.write_bytes(b"\x89PNG\r\n\x1a\n") + with patch("rdt_cli.commands._common.get_credential", return_value=cred): + result = runner.invoke(cli, ["post", "test", "Pic", "--image", str(img), "--draft"]) + assert result.exit_code == 2 + assert "draft" in result.output.lower() + + def test_not_logged_in(self): + with patch("rdt_cli.commands._common.get_credential", return_value=None): + result = runner.invoke(cli, ["post", "test", "Title", "--text", "hi", "--draft"]) + assert result.exit_code == 1 + + +# ── Client post-creation methods (mocked) ─────────────────────────── + + +class TestClientPostMethods: + def _client(self): + from rdt_cli.auth import Credential + from rdt_cli.client import RedditClient + + cred = Credential(cookies={"reddit_session": "x", "csrf_token": "tok"}) + return RedditClient(cred) + + def test_graphql_sends_operation_and_csrf(self): + captured = {} + + def fake_write(method, url, **kwargs): + captured.update(method=method, url=url, json=kwargs.get("json")) + return {"data": {"ok": True}} + + with self._client() as client: + with patch.object(client, "_write_request", side_effect=fake_write): + data = client._graphql("CreatePost", {"input": {"a": 1}}) + assert data == {"ok": True} + assert captured["url"].endswith("/svc/shreddit/graphql") + body = captured["json"] + assert body["operation"] == "CreatePost" + assert body["variables"] == {"input": {"a": 1}} + assert body["csrf_token"] == "tok" + + def test_graphql_raises_on_errors(self): + from rdt_cli.exceptions import RedditApiError + + def fake_write(*a, **k): + return {"data": None, "errors": [{"message": "boom"}]} + + with self._client() as client: + with patch.object(client, "_write_request", side_effect=fake_write): + with pytest.raises(RedditApiError, match="boom"): + client._graphql("CreatePost", {}) + + def test_resolve_subreddit_id(self): + with self._client() as client: + with patch.object( + client, "get_subreddit_about", return_value={"name": "t5_2qh1i", "display_name": "test"} + ): + assert client.resolve_subreddit_id("test") == "t5_2qh1i" + + def test_create_draft_payload(self): + with self._client() as client: + with patch.object(client, "_graphql", return_value={"createDraft": {"ok": True}}) as g: + client.create_draft("t5_x", "Title", body="hello") + op, variables = g.call_args.args + assert op == "CreateDraft" + inp = variables["input"] + assert inp["subredditId"] == "t5_x" + assert inp["title"] == "Title" + assert inp["kind"] == "MARKDOWN" + assert inp["content"] == {"markdown": "hello"} + + def test_create_post_link_payload(self): + with self._client() as client: + with patch.object(client, "_graphql", return_value={}) as g: + client.create_post("t5_x", "T", recaptcha_token="TOK", kind="link", url="https://e.com") + op, variables = g.call_args.args + assert op == "CreatePost" + inp = variables["input"] + assert inp["postType"] == "LINK" + assert inp["url"] == "https://e.com" + assert inp["recaptchaToken"] == "TOK" + assert "correlationId" in inp + + def test_create_profile_post_image_payload(self): + with self._client() as client: + with patch.object(client, "_graphql", return_value={}) as g: + client.create_post( + "t5_prof", "T", recaptcha_token="TOK", kind="image", + media_id="mid42", is_profile=True, + ) + op, variables = g.call_args.args + assert op == "CreateProfilePost" + inp = variables["input"] + assert inp["image"]["url"].endswith("/mid42") + assert inp["recaptchaToken"] == "TOK" + assert "subredditId" not in inp # profile posts omit subredditId + + def test_upload_image(self, tmp_path): + img = tmp_path / "pic.png" + img.write_bytes(b"\x89PNG\r\n\x1a\n") + lease = { + "mediaId": "media123", + "uploadLease": { + "uploadLeaseUrl": "https://s3.example/", + "uploadLeaseHeaders": [{"header": "key", "value": "abc"}], + }, + } + with self._client() as client: + with patch.object(client, "create_media_lease", return_value=lease) as mock_lease: + with patch("httpx.Client.post") as mock_s3: + mock_s3.return_value.status_code = 201 + media_id = client.upload_image(str(img)) + assert media_id == "media123" + mock_lease.assert_called_once_with("PNG") + # S3 upload sends lease fields + file, no reddit cookies + _, kwargs = mock_s3.call_args + assert kwargs["data"] == {"key": "abc"} + assert "file" in kwargs["files"] +