feat: spoof official android app

CHANGELOG.md

@@ -1,3 +1,7 @@
+# Version 0.8.3
+
+- Spoof the official Android app to bypass rate-limiting. Thanks to [redlib](https://github.com/redlib-org/redlib) for the Android app spoofing code.
+
 # Version 0.8.2
 
 - Fix settings cookies expiration as reported by [ValiumBear](https://github.com/ValiumBear) [in](https://github.com/corenting/eddrit/issues/143)

README.md

@@ -2,17 +2,23 @@
 
 ![Build](https://img.shields.io/github/actions/workflow/status/corenting/eddrit/ci.yml?branch=master) ![License](https://img.shields.io/github/license/corenting/eddrit) ![Codecov](https://img.shields.io/codecov/c/github/corenting/eddrit)
 
-An alternative frontend for Reddit, written with Python + [Starlette](https://www.starlette.io/). Inspired by [Nitter](https://github.com/zedeus/nitter), an alternative frontend for Twitter.
+An alternative frontend for Reddit. Inspired by [Nitter](https://github.com/zedeus/nitter), an alternative frontend for Twitter.
+
+Written with Python + [Starlette](https://www.starlette.io/).
 
 - Lightweight
 - No ads
 - Compact design (closer to [old.reddit.com](https://old.reddit.com) than to the redesign)
 - Better mobile support
-- Use the old `.json` API endpoints, no need to register for an OAuth2 identifier for self-hosting
+- No need to register for an OAuth2 identifier for self-hosting: mimic the official Android app by default
 
 Official instance: [eddrit.com](https://eddrit.com)
 
-⚠️ eddrit may get rate-limited by Reddit since they introduced rate-limiting on the API endpoints. In this case, an error message may be displayed.
+## ⚠️ Rate-limiting
+
+By default, eddrit will mimic the official Android app to bypass the rate-limiting (huge thanks to [redlib](https://github.com/redlib-org/redlib) for the implementation).
+
+If you want to use the .json endpoints directly (**you may encounter rate-limiting or be blocked from Reddit**), you can set the environment variable `SPOOFED_CLIENT` (directly or through an environment variable) to `none`.
 
 ## Screenshots
 
@@ -50,4 +56,5 @@ If you wish to support the app, donations are possible [here](https://corenting.
 - [Bootstrap Icons](https://icons.getbootstrap.com/) for the icons used
 - [dash.js](https://github.com/Dash-Industry-Forum/dash.js) for playing videos
 - [Pico.css](https://picocss.com/) for the CSS framework used
+- [redlib](https://github.com/redlib-org/redlib) for the Android app spoofing code
 - [Video.js](https://videojs.com/) for playing videos

eddrit/__init__.py

@@ -4,7 +4,7 @@
 
 from eddrit import config
 
-__version__ = "0.8.2"
+__version__ = "0.8.3"
 
 logger.remove()
 logger.add(sys.stderr, level=config.LOG_LEVEL)

eddrit/app.py

@@ -10,6 +10,7 @@
 from starlette.staticfiles import StaticFiles
 
 from eddrit import __version__, config
+from eddrit.constants import SpoofedClient
 from eddrit.routes.common import exception_handlers
 from eddrit.routes.pages import (
     index,
@@ -21,7 +22,11 @@
     subreddit_and_user,
 )
 from eddrit.routes.xhr import routes
-from eddrit.utils.httpx import raise_if_rate_limited
+from eddrit.utils.api import (
+    event_hook_add_official_android_app_headers_to_request,
+    event_hook_raise_if_rate_limited_on_response,
+    get_official_android_app_headers,
+)
 from eddrit.utils.middlewares import (
     CookiesRefreshMiddleware,
     CurrentHostMiddleware,
@@ -56,12 +61,25 @@ class State(typing.TypedDict):
 
 @contextlib.asynccontextmanager
 async def lifespan(app: Starlette) -> typing.AsyncIterator[State]:
-    """Init the app lifespan with httpx client."""
+    """Init the app lifespan: httpx client etc.."""
+
+    httpx_request_event_hooks = []
+
+    # If spoof client set to official android app, trigger the generation
+    # of the header to avoid blocking on first requests
+    if config.SPOOFED_CLIENT == SpoofedClient.OFFICIAL_ANDROID_APP:
+        get_official_android_app_headers()
+        httpx_request_event_hooks.append(
+            event_hook_add_official_android_app_headers_to_request
+        )
 
     async with httpx.AsyncClient(
         headers={"User-Agent": f"eddrit:v{__version__}"},
         http2=True,
-        event_hooks={"response": [raise_if_rate_limited]},
+        event_hooks={
+            "request": httpx_request_event_hooks,
+            "response": [event_hook_raise_if_rate_limited_on_response],
+        },
     ) as client:
         yield {"http_client": client}
 

eddrit/config.py

@@ -1,7 +1,15 @@
 from environs import Env
 
+from eddrit.constants import SpoofedClient
+
 env = Env()
 env.read_env()
 
-DEBUG: bool = env.bool("DEBUG", False)
-LOG_LEVEL: str = env.str("LOG_LEVEL", "WARNING")
+DEBUG: bool = env.bool("DEBUG", default=False)
+LOG_LEVEL: str = env.str("LOG_LEVEL", default="WARNING")
+SPOOFED_CLIENT: SpoofedClient = env.enum(
+    "SPOOFED_CLIENT",
+    type=SpoofedClient,
+    ignore_case=True,
+    default=SpoofedClient.OFFICIAL_ANDROID_APP.value,
+)

eddrit/constants.py

@@ -0,0 +1,6 @@
+from enum import Enum
+
+
+class SpoofedClient(Enum):
+    NONE = "none"
+    OFFICIAL_ANDROID_APP = "official_android_app"

eddrit/reddit/fetch.py

@@ -5,7 +5,8 @@
 import httpx
 from loguru import logger
 
-from eddrit import models
+from eddrit import config, models
+from eddrit.constants import SpoofedClient
 from eddrit.exceptions import (
     SubredditIsBannedError,
     SubredditIsPrivateError,
@@ -16,6 +17,16 @@
 from eddrit.reddit import parser
 
 
+def get_reddit_base_url() -> str:
+    """
+    Get base URL for call depending on spoof config
+    """
+    if config.SPOOFED_CLIENT == SpoofedClient.OFFICIAL_ANDROID_APP:
+        return "https://oauth.reddit.com"
+
+    return "https://old.reddit.com"
+
+
 async def get_frontpage_information() -> models.Subreddit:
     return models.Subreddit(
         title="Popular",
@@ -32,15 +43,18 @@ async def get_frontpage_posts(
     pagination: models.Pagination,
 ) -> tuple[list[models.Post], models.Pagination]:
     ret = await _get_posts_for_url(
-        http_client, "https://old.reddit.com/.json", pagination, is_popular_or_all=True
+        http_client,
+        f"{get_reddit_base_url()}/.json",
+        pagination,
+        is_popular_or_all=True,
     )
     return ret  # type: ignore
 
 
 async def search_posts(
     http_client: httpx.AsyncClient, input_text: str
 ) -> list[models.Post]:
-    res = await http_client.get(f"https://old.reddit.com/search.json?q={input_text}")
+    res = await http_client.get(f"{get_reddit_base_url()}/search.json?q={input_text}")
     posts, _ = parser.parse_posts_and_comments(res.json(), False)
     # Ignore response type as there is no models.PostComment for search posts
     return posts  # type: ignore
@@ -50,7 +64,7 @@ async def search_subreddits(
     http_client: httpx.AsyncClient, input_text: str
 ) -> list[models.Subreddit]:
     res = await http_client.get(
-        f"https://old.reddit.com/subreddits/search.json?q={input_text}"
+        f"{get_reddit_base_url()}/subreddits/search.json?q={input_text}"
     )
     results = res.json()["data"]["children"]
     return [
@@ -105,22 +119,22 @@ async def get_subreddit_or_user_posts(
 
     if is_user:
         if sorting_mode == models.UserSortingMode.NEW:
-            url = f"https://old.reddit.com/{path_part}/{subreddit_or_username}/.json?t={sorting_period.value}"
+            url = f"{get_reddit_base_url()}/{path_part}/{subreddit_or_username}/.json?t={sorting_period.value}"
         else:
-            url = f"https://old.reddit.com/{path_part}/{subreddit_or_username}/.json?sort={sorting_mode.value}&t={sorting_period.value}"
+            url = f"{get_reddit_base_url()}/{path_part}/{subreddit_or_username}/.json?sort={sorting_mode.value}&t={sorting_period.value}"
     else:
         if sorting_mode == models.SubredditSortingMode.POPULAR:
-            url = f"https://old.reddit.com/{path_part}/{subreddit_or_username}/.json?t={sorting_period.value}"
+            url = f"{get_reddit_base_url()}/{path_part}/{subreddit_or_username}/.json?t={sorting_period.value}"
         else:
-            url = f"https://old.reddit.com/r/{subreddit_or_username}/{sorting_mode.value}.json?t={sorting_period.value}"
+            url = f"{get_reddit_base_url()}/r/{subreddit_or_username}/{sorting_mode.value}.json?t={sorting_period.value}"
 
     return await _get_posts_for_url(http_client, url, pagination)
 
 
 async def get_post(
     http_client: httpx.AsyncClient, subreddit: str, post_id: str
 ) -> models.PostWithComments:
-    url = f"https://old.reddit.com/r/{subreddit}/comments/{post_id}/.json"
+    url = f"{get_reddit_base_url()}/r/{subreddit}/comments/{post_id}/.json"
     params = {"limit": 100}
     res = await http_client.get(url, params=params)
 
@@ -136,7 +150,7 @@ async def get_post(
 async def get_comments(
     http_client: httpx.AsyncClient, subreddit: str, post_id: str, comment_id: str
 ) -> Iterable[models.PostComment | models.PostCommentShowMore]:
-    url = f"https://old.reddit.com/r/{subreddit}/comments/{post_id}/comments/{comment_id}/.json"
+    url = f"{get_reddit_base_url()}/r/{subreddit}/comments/{post_id}/comments/{comment_id}/.json"
     params = {"limit": 100}
     res = await http_client.get(url, params=params)
 
@@ -173,7 +187,7 @@ async def _get_posts_for_url(
 async def get_user_information(
     http_client: httpx.AsyncClient, name: str
 ) -> models.User:
-    res = await http_client.get(f"https://old.reddit.com/user/{name}/about/.json")
+    res = await http_client.get(f"{get_reddit_base_url()}/user/{name}/about/.json")
 
     # If user not found, the API redirects us to search endpoint
     if res.status_code == 404:
@@ -186,7 +200,9 @@ async def get_user_information(
 async def _get_subreddit_information(
     http_client: httpx.AsyncClient, name: str
 ) -> models.Subreddit:
-    res = await http_client.get(f"https://old.reddit.com/r/{name}/about/.json")
+    res = await http_client.get(
+        f"{get_reddit_base_url()}/r/{name}/about/.json?raw_json=1"
+    )
 
     # If subreddit not found, the API redirects us to search endpoint
     if res.status_code == 302 and "search" in res.headers["location"]:
@@ -202,7 +218,7 @@ async def _get_multi_information(
     http_client: httpx.AsyncClient, name: str
 ) -> models.Subreddit:
     # Check if there is a redirect to know if it's an NSFW multi
-    res = await http_client.head(f"https://old.reddit.com/r/{name}")
+    res = await http_client.head(f"{get_reddit_base_url()}/r/{name}")
 
     if len(res.history) > 0 and res.history[0].status_code != 200:
         raise SubredditNotFoundError()