first commit

src/models/github.py

@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class Author(BaseModel):
+    login: str
+    id: int
+    node_id: str
+    avatar_url: str
+    gravatar_id: str
+    url: str
+    html_url: str
+    followers_url: str
+    following_url: str
+    gists_url: str
+    starred_url: str
+    subscriptions_url: str
+    organizations_url: str
+    repos_url: str
+    events_url: str
+    received_events_url: str
+    type: str
+    site_admin: bool
+
+
+class Uploader(BaseModel):
+    login: str
+    id: int
+    node_id: str
+    avatar_url: str
+    gravatar_id: str
+    url: str
+    html_url: str
+    followers_url: str
+    following_url: str
+    gists_url: str
+    starred_url: str
+    subscriptions_url: str
+    organizations_url: str
+    repos_url: str
+    events_url: str
+    received_events_url: str
+    type: str
+    site_admin: bool
+
+
+class Asset(BaseModel):
+    url: str
+    id: int
+    node_id: str
+    name: str
+    label: str | None
+    uploader: Uploader
+    content_type: str
+    state: str
+    size: int
+    download_count: int
+    created_at: str
+    updated_at: str
+    browser_download_url: str
+
+
+class Reactions(BaseModel):
+    url: str
+    total_count: int
+
+    laugh: int
+    hooray: int
+    confused: int
+    heart: int
+    rocket: int
+    eyes: int
+
+
+class GithubResponse(BaseModel):
+    url: str
+    assets_url: str
+    upload_url: str
+    html_url: str
+    id: int
+    author: Author
+    node_id: str
+    tag_name: str
+    target_commitish: str
+    name: str
+    draft: bool
+    prerelease: bool
+    created_at: str
+    published_at: str
+    assets: list[Asset]
+    tarball_url: str
+    zipball_url: str
+    body: str
+    reactions: Reactions

src/models/requests.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+import re
+import time
+from typing import Any
+
+from nodriver import Tab
+from pydantic import BaseModel
+
+
+class LinkRequest(BaseModel):
+    cmd: str
+    url: str
+    maxTimeout: int  # noqa: N815 # Ignore to preserve compatibility
+
+
+class ProtectionTriggeredError(Exception):
+    pass
+
+
+class Solution(BaseModel):
+    url: str
+    status: int
+    cookies: list[dict]
+    userAgent: str  # noqa: N815 # Ignore to preserve compatibility
+    headers: dict[str, Any]
+    response: str
+
+
+class LinkResponse(BaseModel):
+    status: str = "ok"
+    message: str
+    solution: Solution
+    startTimestamp: int  # noqa: N815 # Ignore to preserve compatibility
+    endTimestamp: int = int(time.time() * 1000)  # noqa: N815 # Ignore to preserve compatibility
+    version: str = "3.3.21"  # TODO: Implement versioning
+
+    @classmethod
+    async def create(
+        cls,
+        page: Tab,
+        start_timestamp: int,
+        *,
+        challenged: bool = False,
+    ):
+        message = "Passed challenge" if challenged else "Challenge not detected"
+
+        user_agent = await page.js_dumps("navigator")
+        if not isinstance(user_agent, dict):
+            raise ProtectionTriggeredError("User agent is not a dictionary")
+        user_agent = user_agent["userAgent"]
+        re.sub(pattern="HEADLESS", repl="", string=user_agent, flags=re.IGNORECASE)
+
+        # cookies = await page.browser.cookies.get_all(requests_cookie_format=True)
+        # # Convert cookies to json
+        # cookies = [cookie.to_json() for cookie in cookies]
+
+        cookies = []
+        solution = Solution(
+            url=page.url,
+            status=200,
+            cookies=cookies,
+            userAgent=user_agent,
+            headers={},
+            response=await page.get_content(),
+        )
+
+        return cls(
+            message=message,
+            solution=solution,
+            startTimestamp=start_timestamp,
+        )

src/utils/__init__.py

@@ -0,0 +1,3 @@
+import logging
+
+logger = logging.getLogger("uvicorn.error")

src/utils/consts.py

@@ -0,0 +1,13 @@
+CHALLENGE_TITLES = [
+    # Cloudflare
+    "Just a moment...",
+    # DDoS-GUARD
+    "DDoS-Guard",
+]
+
+EXTENTION_REPOSITIORIES = [
+    "OhMyGuus/I-Still-Dont-Care-About-Cookies",
+    "uBlockOrigin/uBOL-home",
+]
+
+SLEEP_SECONDS = 1

src/utils/extentions.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from pathlib import Path
+from zipfile import ZipFile
+
+import httpx
+
+from src.models.github import GithubResponse
+from src.utils.consts import EXTENTION_REPOSITIORIES
+
+downloaded_extentions: list[str] = []  # Do not modify, use download_extentions()
+
+
+def get_latest_github_release(url: str):
+    """Get the latest release from GitHub."""
+    url = "https://api.github.com/repos/" + url
+    url += "/releases/latest"
+    response = GithubResponse.model_validate_json(httpx.get(url).text)
+
+    for asset in response.assets:
+        if "chrom" in asset.name:
+            return asset
+
+    raise ValueError("Couldn't find chrome verions of the release")
+
+
+def download_extentions():
+    """Download the extention."""
+    for repository in EXTENTION_REPOSITIORIES:
+        extention = get_latest_github_release(repository)
+        extention_name = repository.split("/")[-1]
+        response = httpx.get(extention.browser_download_url)
+        Path("extentions").mkdir(exist_ok=True)
+        if not Path(f"extentions/{extention.name}").exists():
+            with Path(f"extentions/{extention.name}").open("wb") as f:
+                f.write(response.content)
+
+            with ZipFile(f"extentions/{extention.name}", "r") as zip_obj:
+                zip_obj.extractall(f"extentions/{extention_name}")
+
+        downloaded_extentions.append(f"extentions/{extention_name}")

src/utils/misc.py

@@ -0,0 +1,27 @@
+import nodriver as webdriver
+from nodriver.core.element import Element
+
+from src.utils import logger
+from src.utils.utils import CHALLENGE_TITLES
+
+
+async def new_browser():
+    config: webdriver.Config = webdriver.Config()
+    config.sandbox = False
+
+    return await webdriver.start(config=config)
+
+
+async def bypass_cloudflare(page: webdriver.Tab):
+    challenged = False
+    while True:
+        logger.info("Bypassing cloudflare")
+        await page
+        await page.wait(0.5)
+        if page.target.title not in CHALLENGE_TITLES:
+            return challenged
+        challenged = True
+        elem = await page.query_selector(".cf-turnstile-wrapper")
+        if isinstance(elem, Element):
+            logger.info(f"Clicking element: {elem}")
+            await elem.mouse_click()

src/utils/utils.py

@@ -0,0 +1,29 @@
+import logging
+from venv import logger
+
+import nodriver as webdriver
+from nodriver.core.element import Element
+
+from src.utils.consts import CHALLENGE_TITLES
+
+
+def get_logger():
+    return logging.getLogger("uvicorn.error")
+
+
+async def new_browser():
+    config: webdriver.Config = webdriver.Config()
+    config.sandbox = False
+
+    return await webdriver.start(config=config)
+
+
+async def bypass_cloudflare(page: webdriver.Tab):
+    while True:
+        await page
+        if page.target.title not in CHALLENGE_TITLES:
+            break
+        elem = await page.query_selector(".cf-turnstile-wrapper")
+        if isinstance(elem, Element):
+            logger.info(f"Clicking element: {elem}")
+            await elem.mouse_click()