Raspberry Pi＋WordPressでX自動告知

この記事は固有情報を隠すため、<PI_HOST>, <PI_USER>, https://example.com などプレースホルダで書いています。コピペ後に自分の値へ置換してください。

Xへの自動投稿を“ちゃんと”作る

― 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

１. 全体図

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

最小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!"))

画像・動画を添付する

現行の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
  • 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
  • サーバ側（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などの再試行は指数バックオフで。

エラーの読み方（実例）

料金・上限の現実的メモ

月間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

公式ドキュメント（抜粋）

ライセンス：本記事のテキスト/コードは特記なき限り CC BY 4.0 です。引用の際は出典URL（本ページ）を明記してください。
利用ポリシー もご参照ください。

コメント

タイトルとURLをコピーしました