この記事は固有情報を隠すため、
<PI_HOST>,<PI_USER>,https://example.comなどプレースホルダで書いています。コピペ後に自分の値へ置換してください。
Xへの自動投稿を“ちゃんと”作る
- ― OAuth 2.0 PKCE、127.0.0.1コールバック、トークン管理、メディア添付、SSHトンネルまで
- 1. 全体図
- なぜ 127.0.0.1 で、localhost はダメなの?
- OAuth 2.0(PKCE)を“正しく”通す
- 最小Python実装(PKCE+ローカルコールバック)
- 実際に投稿する(テキストのみ)
- 画像・動画を添付する
- 失敗しないApp設定の要点(エビデンス付きチェックリスト)
- ネットワーク/プロトコル:SSHローカルフォワードで“Windows → Linux(Raspberry Pi)”を繋ぐ
- 文字数制限・スレッド投稿・重複防止
- エラーの読み方(実例)
- 料金・上限の現実的メモ
- さらなるハードニング(運用Tips)
- 参考:メディア添付の最小コード(抜粋)
- 付録:HTTPリダイレクトの仕組み(1分で理解)
― OAuth 2.0 PKCE、127.0.0.1コールバック、トークン管理、メディア添付、SSHトンネルまで
- 投稿だけなら必要スコープは
tweet.write+users.read(多くのエンドポイントでtweet.readも併記) X Developer Platform - 認可は OAuth 2.0(Authorization Code with PKCE)。認可URLは
https://x.com/i/oauth2/authorize、トークンはhttps://api.x.com/2/oauth2/token。 - コールバックに
localhostは使わない。代わりにhttp(s)://127.0.0.1を使う(App設定にも同じURLを許可登録)。 X Developer PlatformDeveloper X - 投稿は
POST https://api.x.com/2/tweets。Bearerでユーザーアクセストークンを送る。 X Developer Platform - メディアは v2 の
/2/media/upload(INIT/APPEND/FINALIZE) でアップロード→media_idsをPOST /2/tweetsに添付。 X Developer Platform - トークンは短命(
expires_in)。offline.accessを付けてリフレッシュトークンを貰って自動更新しよう。 X Developer PlatformOAuth 2.0 Simplified - TLS 1.2必須。クライアントのルートストア更新も忘れずに。 X Developer Platform
1. 全体図
flowchart LR
subgraph Client["自動投稿クライアント(Python)"]
A[PKCE生成] --> B[認可URL生成]
B -->|ブラウザ起動| C[ユーザログイン]
E[127.0.0.1:8000 ローカルHTTPサーバ] --> F[認可コード受領]
F --> G[トークン交換 /2/oauth2/token]
G --> H[投稿 /2/tweets]
H --> I[結果保存・再試行制御]
end
subgraph X["X Platform"]
XAuth[Auth Server]:::cloud
XAPI[REST API]:::cloud
end
C-->XAuth
E<--XAuth
G-->XAuth
H-->XAPI
classDef cloud fill:#eef,stroke:#99f
なぜ 127.0.0.1 で、localhost はダメなの?
Xの公式ガイダンスに**「localhostはコールバックURLに使わないで。代わりに http(s)://127.0.0.1 を使って」と明記されています。さらにApp設定(許可リスト)と実際に投げる redirect_uri は完全一致**である必要があります。 X Developer PlatformDeveloper X
ネットワーク的背景:
localhostはOSの名前解決に依存し、開発環境によっては::1(IPv6)へ解決され、ブラウザ↔ローカルサーバ間でアドレス不一致が起こりがち。127.0.0.1はIPv4のループバック(OSカーネル内で折り返す)で必ず自ホストに到達します。IPv4ループバックは 127.0.0.0/8 全体が予約されています。 IETF Datatracker+1
OAuth 2.0(PKCE)を“正しく”通す
プロトコルの流れ
sequenceDiagram autonumber participant App as 自動投稿アプリ participant Browser as ブラウザ participant XAuth as X 認可サーバ participant Local as 127.0.0.1:8000 App->>App: code_verifier生成(長さ43〜128 / RFC7636) App->>App: code_challenge=S256(code_verifier) App->>Browser: https://x.com/i/oauth2/authorize?...&code_challenge=... Browser->>XAuth: 認証+同意(tweet.write, users.read, offline.access) XAuth-->>Local: 302リダイレクト(?code=...&state=...) Local->>App: 認可コードを渡す App->>XAuth: POST /2/oauth2/token(code, code_verifier) XAuth-->>App: access_token(短命), refresh_token App->>XAuth: POST /2/oauth2/token(refresh_token)※期限切れ時 XAuth-->>App: 新しいaccess_token
- PKCEの要件:
code_verifierは43〜128文字、code_challengeは通常S256推奨。 IETF Datatracker - エンドポイント:認可
x.com/i/oauth2/authorize、トークンapi.x.com/2/oauth2/token。 X Developer Platform - TLS 1.2必須。 X Developer Platform
- トークン寿命:
expires_inを確認。offline.accessでリフレッシュトークンを取得し、無人更新するのが定石。 X Developer PlatformOAuth 2.0 Simplified
最小Python実装(PKCE+ローカルコールバック)
依存:
requests、urllib3(標準http.server使用)。
App設定:OAuth2を有効化し、スコープtweet.write users.read offline.access、Callbackにhttp://127.0.0.1:8000/callbackを登録。 X Developer Platform
# x_oauth_pkce_min.py
import base64, hashlib, os, secrets, json, threading, webbrowser
from http.server import BaseHTTPRequestHandler, HTTPServer
import urllib.parse as up
import requests
from pathlib import Path
from time import time
CLIENT_ID = os.getenv("X_CLIENT_ID") # developer portalのClient ID
REDIRECT_URI = "http://127.0.0.1:8000/callback"
SCOPES = "tweet.read users.read tweet.write offline.access"
AUTH_URL = "https://x.com/i/oauth2/authorize"
TOKEN_URL = "https://api.x.com/2/oauth2/token"
TOK_PATH = Path("x_tokens.json")
def b64urle(b: bytes)->str:
return base64.urlsafe_b64encode(b).rstrip(b"=").decode()
def new_pkce():
verifier = b64urle(secrets.token_bytes(64)) # 86文字程度、RFC7636要件内
challenge = b64urle(hashlib.sha256(verifier.encode()).digest())
return verifier, challenge
class CodeHandler(BaseHTTPRequestHandler):
server_version = "MiniAuth/1.0"
code = state = None
def do_GET(self):
q = up.urlparse(self.path)
if q.path != "/callback":
self.send_response(404); self.end_headers(); return
params = up.parse_qs(q.query)
CodeHandler.code = params.get("code", [None])[0]
CodeHandler.state = params.get("state", [None])[0]
self.send_response(200); self.end_headers()
self.wfile.write(b"OK. You can close this tab.")
def log_message(self, *_): pass
def run_local_server():
httpd = HTTPServer(("127.0.0.1", 8000), CodeHandler)
threading.Thread(target=httpd.serve_forever, daemon=True).start()
return httpd
def save_tokens(data):
data["obtained_at"] = int(time())
TOK_PATH.write_text(json.dumps(data, indent=2))
def load_tokens():
if TOK_PATH.exists():
return json.loads(TOK_PATH.read_text())
return None
def token_expired(t):
# expires_in(秒)を見て、余裕をもって更新
return not t or (time() - t.get("obtained_at", 0)) > (t.get("expires_in", 0) - 60)
def exchange_code(code, verifier):
d = {
"grant_type":"authorization_code",
"client_id": CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"code": code,
"code_verifier": verifier,
}
r = requests.post(TOKEN_URL, data=d, timeout=20)
r.raise_for_status()
return r.json()
def refresh(refresh_token):
d = {
"grant_type":"refresh_token",
"client_id": CLIENT_ID,
"refresh_token": refresh_token,
}
r = requests.post(TOKEN_URL, data=d, timeout=20)
r.raise_for_status()
return r.json()
def ensure_token():
t = load_tokens()
if t and not token_expired(t):
return t["access_token"]
if t and "refresh_token" in t:
nt = refresh(t["refresh_token"])
save_tokens(nt)
return nt["access_token"]
verifier, challenge = new_pkce()
state = b64urle(secrets.token_bytes(16))
params = {
"response_type":"code",
"client_id": CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"scope": SCOPES,
"state": state,
"code_challenge": challenge,
"code_challenge_method":"S256",
}
url = AUTH_URL + "?" + up.urlencode(params, quote_via=up.quote)
httpd = run_local_server()
webbrowser.open(url)
# wait until handler sets code
import time as _t
for _ in range(600):
if CodeHandler.code:
break
_t.sleep(0.2)
httpd.shutdown()
if not CodeHandler.code:
raise RuntimeError("No code received")
tok = exchange_code(CodeHandler.code, verifier)
save_tokens(tok)
return tok["access_token"]
if __name__ == "__main__":
print("Access token:", ensure_token()[:16] + "...")
- 認可URL/トークンURL/スコープ仕様は公式のまま。 X Developer Platform+1
expires_inを見て先行リフレッシュ。offline.accessがなければリフレッシュトークンは出ません。 X Developer Platform
実際に投稿する(テキストのみ)
# post_text.py
import requests, json
from x_oauth_pkce_min import ensure_token
API_POST = "https://api.x.com/2/tweets"
def post_text(text: str):
tok = ensure_token()
headers = {"Authorization": f"Bearer {tok}", "Content-Type":"application/json"}
body = {"text": text}
r = requests.post(API_POST, headers=headers, data=json.dumps(body), timeout=20)
r.raise_for_status()
return r.json()
if __name__ == "__main__":
print(post_text("Hello from API v2!"))
- エンドポイント:
POST /2/tweets。レスポンス201が成功。 X Developer Platform - 要スコープ:
tweet.write(+users.read/tweet.read)。 X Developer Platform
画像・動画を添付する
現行のv2では /2/media/upload のチャンクアップロード(INIT→APPEND→FINALIZE) が利用可能。アップロード後の media_id を POST /2/tweets に添付 します。 X Developer Platform
sequenceDiagram
participant App
participant Media as /2/media/upload
App->>Media: INIT (total_bytes, media_type)
loop each chunk
App->>Media: APPEND (segment_index, chunk)
end
App->>Media: FINALIZE
Media-->>App: media_id
App->>/2/tweets: media_ids: [media_id]注意:一部の古い統合ガイドは「v2でフルアップロード不可」と書いていますが、現在はv2のメディアアップロードが案内されています(最新ドキュメント参照)。 X Developer Platform+1
失敗しないApp設定の要点(エビデンス付きチェックリスト)
- Callback URLs
- App設定の「Callback URL許可リスト」に
http://127.0.0.1:8000/callbackを登録。 - 認可リクエストの
redirect_uriと完全一致させる。 localhostは使わない。 X Developer PlatformDeveloper X
- App設定の「Callback URL許可リスト」に
- OAuth 2.0(PKCE)を選択し、スコープに
tweet.write users.read(必要に応じてtweet.read)+無人運用ならoffline.access。 X Developer Platform+1 - TLS 1.2必須(クライアントのルートストア更新も)。 X Developer Platform
ネットワーク/プロトコル:SSHローカルフォワードで“Windows → Linux(Raspberry Pi)”を繋ぐ
「Windowsでブラウザ認証 → 127.0.0.1:8000に返ってくる → そのポートをPi上のPythonに転送したい」ケースを想定。
graph LR
B[Windows ブラウザ] -- https://x.com/i/oauth2/authorize --> X[x.com]
X -- 302 --> CB[Windows 127.0.0.1:8000]
subgraph SSH Tunnel
CB ==ssh -L 8000:127.0.0.1:8000==> PiL[Pi 127.0.0.1:8000]
end
PiL --> AppPi[Pi: ローカルHTTPサーバ&投稿スクリプト]ssh -L [local_addr:]LPORT:DEST:DESTPORT HOSTは「クライアント側ポート→(暗号化)→サーバ側の任意ホスト:ポートへ中継」。- 例(Windows PowerShell):
ssh -N -L 127.0.0.1:8000:127.0.0.1:8000 pi@raspi.local -Nはリモートコマンド実行なしでトンネル専用。OpenSSHのローカルフォワーディング仕様。 man.openbsd.org
- 例(Windows PowerShell):
- サーバ側(Pi)の
sshd_configでAllowTcpForwarding yesが必要なことがある。 man.openbsd.orgman7.org - ループバックの性質(127.0.0.0/8は転送されず必ず自機内で完結)を理解しておくとデバッグが速い。 IETF Datatracker
補足(Windowsの常駐化)
- ずっと張りたいならタスクスケジューラで
ssh -N -L ...をログオン時起動に。切断時の再接続は-o ServerAliveInterval=30 -o ServerAliveCountMax=3を併用。
文字数制限・スレッド投稿・重複防止
- Xの投稿本文の扱いは v2の「数え方」ドキュメントを参照(絵文字、URL短縮など)。 X Developer Platform
- 重複防止は、本文のハッシュ+時刻で冪等性キーを自前管理(例:直近N件のsha256をKVに保持)。429/503などの再試行は指数バックオフで。
エラーの読み方(実例)
invalid_request(OAuth2):redirect_uri不一致など。 X Developer PlatformCallback URL not approved...:Appの許可リストに未登録。 X Developer Platform- 401/403(投稿時):スコープ不足・失効トークン。
expires_inを見て更新、スコープを追加してユーザーに再同意。 X Developer PlatformOAuth 2.0 Simplified
料金・上限の現実的メモ
月間Post上限(“Post cap”)はプランで異なります。 無人運用では上限到達時の挙動(停止 or 翌月まで延期)を設計しましょう。最新のプランと制限は公式のプラン/Post capページで必ず確認してください。 X Developer Platform+1
さらなるハードニング(運用Tips)
- トークン保管:平文JSONではなく、OSの秘匿ストアやKMSを使用。Bearerは流出=即不正利用(RFC6750)。 IETF Datatracker
- HTTP→TLS:コールバック受け取りはローカルでも、外向き通信は当然TLS(v1.2+)。 X Developer Platform
- ログ:
stateの突合、x-access-level、HTTPステータスとレスポンス本文をサニタイズして保存。 X Developer Platform
参考:メディア添付の最小コード(抜粋)
# media_upload_min.py
import requests, json, math
from x_oauth_pkce_min import ensure_token
MEDIA_URL = "https://api.x.com/2/media/upload"
POST_URL = "https://api.x.com/2/tweets"
def upload_media(path, media_type="image/png", chunk=5*1024*1024):
tok = ensure_token()
h = {"Authorization": f"Bearer {tok}"}
size = os.path.getsize(path)
# INIT
r = requests.post(MEDIA_URL, headers=h, data={"command":"INIT","total_bytes":size,"media_type":media_type})
r.raise_for_status()
media_id = r.json()["media_id"]
# APPEND
with open(path,"rb") as f:
idx=0
while True:
buf = f.read(chunk)
if not buf: break
files = {"media": buf}
d = {"command":"APPEND","media_id":media_id,"segment_index":idx}
rr = requests.post(MEDIA_URL, headers=h, data=d, files=files)
rr.raise_for_status()
idx+=1
# FINALIZE
r = requests.post(MEDIA_URL, headers=h, data={"command":"FINALIZE","media_id":media_id})
r.raise_for_status()
return media_id
def post_with_media(text, media_path):
mid = upload_media(media_path)
tok = ensure_token()
headers={"Authorization": f"Bearer {tok}", "Content-Type":"application/json"}
body={"text": text, "media":{"media_ids":[mid]}}
r = requests.post(POST_URL, headers=headers, data=json.dumps(body))
r.raise_for_status()
return r.json()
- チャンク型アップロード(INIT/APPEND/FINALIZE)は公式クイックスタートに準拠。 X Developer Platform
付録:HTTPリダイレクトの仕組み(1分で理解)
sequenceDiagram participant Browser participant XAuth participant App Browser->>XAuth: GET /i/oauth2/authorize?... XAuth-->>Browser: 302 Found + Location: http://127.0.0.1:8000/callback?code=... Browser->>App: GET /callback?code=...&state=...
- ブラウザは
Locationヘッダへ移動(HTTP 302)。ローカルのHTTPサーバがコードを受領→アプリがトークンと交換。 - X APIとのすべてのやり取りはTLS 1.2で保護される。 X Developer Platform
公式ドキュメント(抜粋)
- 認可〜トークン(PKCE):
/i/oauth2/authorize、/2/oauth2/tokenX Developer Platform - スコープとエンドポイント対応表(投稿に必要スコープ):X Developer Platform
- 投稿API:
POST /2/tweetsX Developer Platform - メディアアップロード(v2):
/2/media/upload(INIT/APPEND/FINALIZE) X Developer Platform - Callback/localhost禁止・許可リスト・完全一致: X Developer PlatformDeveloper X
- TLS要件(1.2必須): X Developer Platform
- 127.0.0.0/8はループバック: IETF Datatracker

コメント