add the get_account.py script from upstream

Signed-off-by: ngn <ngn@ngn.tf>

.gitea/workflows/build.yml

@@ -1,4 +1,4 @@
-name: Build and publish the docker image
+name: Build and publish the docker images
 
 on:
   push:
@@ -22,7 +22,12 @@ jobs:
           username: ${{gitea.actor}}
           password: ${{secrets.PACKAGES_TOKEN}}
 
-      - name: Build image
+      - name: Build images
+        run: |
+          docker build . -f docker/nitter/Dockerfile --tag ${{env.REGISTRY}}/${{env.IMAGE}}:latest
+          docker build . -f docker/get_account/Dockerfile --tag ${{env.REGISTRY}}/${{env.IMAGE}}-get-account:latest
+
+      - name: Push images
         run: |
-          docker build . --tag ${{env.REGISTRY}}/${{env.IMAGE}}:latest
           docker push ${{env.REGISTRY}}/${{env.IMAGE}}:latest
+          docker push ${{env.REGISTRY}}/${{env.IMAGE}}-get-account:latest

.gitignore

@@ -1,4 +1,4 @@
-nitter
+./nitter
 *.html
 *.db
 data

docker/get_account/Dockerfile

@@ -0,0 +1,6 @@
+FROM python
+
+RUN pip install pyotp requests
+COPY ./tools/get_account.py /get_account.py
+
+ENTRYPOINT ["python3", "/get_account.py"]

scripts/oauth.sh

@@ -1,52 +0,0 @@
-#!/bin/bash -e
-
-# Grab oauth token for use with Nitter (requires Twitter account).
-# results: {"oauth_token":"xxxxxxxxxx-xxxxxxxxx","oauth_token_secret":"xxxxxxxxxxxxxxxxxxxxx"}
-
-if [ $# -ne 2 ]; then
-  echo "please specify a username and password"
-  exit 1
-fi
-
-username="${1}"
-password="${2}"
-
-bearer_token='AAAAAAAAAAAAAAAAAAAAAFXzAwAAAAAAMHCxpeSDG1gLNLghVe8d74hl6k4%3DRUMF4xAQLsbeBhTSRrCiQpJtxoGWeyHrDb5te2jpGskWDFW82F'
-guest_token=$(curl -s -XPOST https://api.twitter.com/1.1/guest/activate.json -H "Authorization: Bearer ${bearer_token}" | jq -r '.guest_token')
-base_url='https://api.twitter.com/1.1/onboarding/task.json'
-header=(-H "Authorization: Bearer ${bearer_token}" -H "User-Agent: TwitterAndroid/10.21.1" -H "Content-Type: application/json" -H "X-Guest-Token: ${guest_token}")
-
-# start flow
-flow_1=$(curl -si -XPOST "${base_url}?flow_name=login" "${header[@]}")
-
-# get 'att', now needed in headers, and 'flow_token' from flow_1
-att=$(sed -En 's/^att: (.*)\r/\1/p' <<< "${flow_1}")
-flow_token=$(sed -n '$p' <<< "${flow_1}" | jq -r .flow_token)
-
-if [[ -z "$flow_1" || -z "$flow_token" ]]; then
-  echo "Couldn't retrieve flow token (twitter not reachable?)"
-  exit 1
-fi
-
-# username
-token_2=$(curl -s -XPOST "${base_url}" -H "att: ${att}" "${header[@]}" \
-  -d '{"flow_token":"'"${flow_token}"'","subtask_inputs":[{"subtask_id":"LoginEnterUserIdentifierSSO","settings_list":{"setting_responses":[{"key":"user_identifier","response_data":{"text_data":{"result":"'"${username}"'"}}}],"link":"next_link"}}]}' | jq -r .flow_token)
-
-if [[ -z "$token_2" || "$token_2" == "null" ]]; then
-  echo "Couldn't retrieve user token (check if login is correct)"
-  exit 1
-fi
-
-# password
-token_3=$(curl -s -XPOST "${base_url}" -H "att: ${att}" "${header[@]}" \
-  -d '{"flow_token":"'"${token_2}"'","subtask_inputs":[{"enter_password":{"password":"'"${password}"'","link":"next_link"},"subtask_id":"LoginEnterPassword"}]}' | jq -r .flow_token)
-
-if [[ -z "$token_3" || "$token_3" == "null" ]]; then
-  echo "Couldn't retrieve user token (check if password is correct)"
-  exit 1
-fi
-
-# finally print oauth_token and secret
-curl -s -XPOST "${base_url}" -H "att: ${att}" "${header[@]}" \
-  -d '{"flow_token":"'"${token_3}"'","subtask_inputs":[{"check_logged_in_account":{"link":"AccountDuplicationCheck_false"},"subtask_id":"AccountDuplicationCheck"}]}' | \
-  jq -c '.subtasks[0]|if(.open_account) then [{oauth_token: .open_account.oauth_token, oauth_token_secret: .open_account.oauth_token_secret}] else empty end'

tools/get_account.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python3
+from sys import argv
+import requests
+import pyotp
+import json
+
+TW_CONSUMER_KEY = "3nVuSoBZnx6U4vzUxf5w"
+TW_CONSUMER_SECRET = "Bcs59EFbbsdF6Sl9Ng71smgStWEGwXXKSjYvPVt7qys"
+
+
+def auth(username, password, otp_secret):
+    bearer_token_req = requests.post(
+        "https://api.twitter.com/oauth2/token",
+        auth=(TW_CONSUMER_KEY, TW_CONSUMER_SECRET),
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+        data="grant_type=client_credentials",
+    ).json()
+    bearer_token = " ".join(str(x) for x in bearer_token_req.values())
+
+    guest_token = (
+        requests.post(
+            "https://api.twitter.com/1.1/guest/activate.json",
+            headers={"Authorization": bearer_token},
+        )
+        .json()
+        .get("guest_token")
+    )
+
+    if not guest_token:
+        print("Failed to obtain guest token.")
+        exit(1)
+
+    twitter_header = {
+        "Authorization": bearer_token,
+        "Content-Type": "application/json",
+        "User-Agent": "TwitterAndroid/10.21.0-release.0 (310210000-r-0) ONEPLUS+A3010/9 (OnePlus;ONEPLUS+A3010;OnePlus;OnePlus3;0;;1;2016)",
+        "X-Twitter-API-Version": "5",
+        "X-Twitter-Client": "TwitterAndroid",
+        "X-Twitter-Client-Version": "10.21.0-release.0",
+        "OS-Version": "28",
+        "System-User-Agent": "Dalvik/2.1.0 (Linux; U; Android 9; ONEPLUS A3010 Build/PKQ1.181203.001)",
+        "X-Twitter-Active-User": "yes",
+        "X-Guest-Token": guest_token,
+        "X-Twitter-Client-DeviceID": "",
+    }
+
+    session = requests.Session()
+    session.headers = twitter_header
+
+    task1 = session.post(
+        "https://api.twitter.com/1.1/onboarding/task.json",
+        params={
+            "flow_name": "login",
+            "api_version": "1",
+            "known_device_token": "",
+            "sim_country_code": "us",
+        },
+        json={
+            "flow_token": None,
+            "input_flow_data": {
+                "country_code": None,
+                "flow_context": {
+                    "referrer_context": {
+                        "referral_details": "utm_source=google-play&utm_medium=organic",
+                        "referrer_url": "",
+                    },
+                    "start_location": {"location": "deeplink"},
+                },
+                "requested_variant": None,
+                "target_user_id": 0,
+            },
+        },
+    )
+
+    session.headers["att"] = task1.headers.get("att")
+
+    task2 = session.post(
+        "https://api.twitter.com/1.1/onboarding/task.json",
+        json={
+            "flow_token": task1.json().get("flow_token"),
+            "subtask_inputs": [
+                {
+                    "enter_text": {
+                        "suggestion_id": None,
+                        "text": username,
+                        "link": "next_link",
+                    },
+                    "subtask_id": "LoginEnterUserIdentifier",
+                }
+            ],
+        },
+    )
+
+    task3 = session.post(
+        "https://api.twitter.com/1.1/onboarding/task.json",
+        json={
+            "flow_token": task2.json().get("flow_token"),
+            "subtask_inputs": [
+                {
+                    "enter_password": {"password": password, "link": "next_link"},
+                    "subtask_id": "LoginEnterPassword",
+                }
+            ],
+        },
+    )
+
+    for t3_subtask in task3.json().get("subtasks", []):
+        if "open_account" in t3_subtask:
+            return t3_subtask["open_account"]
+        elif "enter_text" in t3_subtask:
+            response_text = t3_subtask["enter_text"]["hint_text"]
+            totp = pyotp.TOTP(otp_secret)
+            generated_code = totp.now()
+            task4resp = session.post(
+                "https://api.twitter.com/1.1/onboarding/task.json",
+                json={
+                    "flow_token": task3.json().get("flow_token"),
+                    "subtask_inputs": [
+                        {
+                            "enter_text": {
+                                "suggestion_id": None,
+                                "text": generated_code,
+                                "link": "next_link",
+                            },
+                            "subtask_id": "LoginTwoFactorAuthChallenge",
+                        }
+                    ],
+                },
+            )
+            task4 = task4resp.json()
+            for t4_subtask in task4.get("subtasks", []):
+                if "open_account" in t4_subtask:
+                    return t4_subtask["open_account"]
+
+    return None
+
+
+if __name__ == "__main__":
+    if len(argv) != 4:
+        print("Usage: %s <username> <password> <2fa secret>" % argv[0])
+        exit(1)
+
+    username = argv[1]
+    password = argv[2]
+    otp_secret = argv[3]
+
+    result = auth(username, password, otp_secret)
+    if result is None:
+        print("Authentication failed.")
+        exit(1)
+
+    print(
+        json.dumps(
+            {
+                "oauth_token": result.get("oauth_token"),
+                "oauth_token_secret": result.get("oauth_token_secret"),
+            }
+        )
+    )