""" e-Gov 電子申請 API v2 クライアント OAuth2 + PKCE フローによる認証と電子送達ファイルのダウンロード """ import hashlib import base64 import secrets import os import json import time from pathlib import Path from typing import Optional import httpx # ── エンドポイント定数 ───────────────────────────────────────── # EGOV_ENV=sandbox で検証環境、省略または production で本番環境を使用 _ENV = os.environ.get("EGOV_ENV", "production").lower() if _ENV == "sandbox": AUTH_BASE = "https://account2.sbx.e-gov.go.jp/auth" API_BASE = "https://api2.sbx.e-gov.go.jp/shinsei/v2" else: AUTH_BASE = "https://account.e-gov.go.jp/auth" API_BASE = "https://api.e-gov.go.jp/shinsei/v2" AUTH_URL = f"{AUTH_BASE}/auth" TOKEN_URL = f"{AUTH_BASE}/token" LOGOUT_URL = f"{AUTH_BASE}/logout" EGOV_DATA_DIR = Path("/app/egov_data") TOKENS_PATH = EGOV_DATA_DIR / "tokens.json" CONFIG_PATH = EGOV_DATA_DIR / "api_config.json" # ── 資格情報 ────────────────────────────────────────────────── def load_api_config() -> dict: """API設定を config ファイルから読み込む（環境変数でも上書き可）""" cfg = {} if CONFIG_PATH.exists(): try: with open(CONFIG_PATH) as f: cfg = json.load(f) except Exception: pass # 環境変数が優先 client_id = os.environ.get("EGOV_SOFTWARE_ID") or cfg.get("software_id", "") client_secret = os.environ.get("EGOV_API_KEY") or cfg.get("api_key", "") redirect_uri = os.environ.get("EGOV_REDIRECT_URI") or cfg.get("redirect_uri", "http://192.168.0.61:18000/egov.html") return { "client_id": client_id, "client_secret": client_secret, "redirect_uri": redirect_uri, } def save_api_config(software_id: str, api_key: str, redirect_uri: str = "http://192.168.0.61:18000/egov.html"): """API設定をファイルに保存""" EGOV_DATA_DIR.mkdir(parents=True, exist_ok=True) with open(CONFIG_PATH, "w") as f: json.dump({"software_id": software_id, "api_key": api_key, "redirect_uri": redirect_uri}, f) # ── トークン永続化 ──────────────────────────────────────────── def load_tokens() -> dict: if TOKENS_PATH.exists(): try: with open(TOKENS_PATH) as f: return json.load(f) except Exception: pass return {} def save_tokens(tokens: dict): EGOV_DATA_DIR.mkdir(parents=True, exist_ok=True) with open(TOKENS_PATH, "w") as f: json.dump(tokens, f) def delete_tokens(): if TOKENS_PATH.exists(): TOKENS_PATH.unlink() # ── PKCE ───────────────────────────────────────────────────── def generate_pkce() -> tuple[str, str]: """code_verifier と code_challenge (S256) を生成""" verifier = secrets.token_urlsafe(48) # 64文字 (43-128の範囲内) digest = hashlib.sha256(verifier.encode()).digest() challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode() return verifier, challenge # ── OAuth2 フロー ───────────────────────────────────────────── def build_auth_url(client_id: str, redirect_uri: str, challenge: str, state: str) -> str: """OAuth2 認可URL を生成""" from urllib.parse import urlencode, quote params = { "client_id": client_id, "response_type": "code", "scope": "openid offline_access", "redirect_uri": redirect_uri, "state": state, "code_challenge": challenge, "code_challenge_method": "S256", } # quote_via=quote を使うことでスペースを + ではなく %20 にエンコードする return f"{AUTH_URL}?" + urlencode(params, quote_via=quote) async def exchange_code(code: str, client_id: str, client_secret: str, redirect_uri: str, verifier: str) -> dict: """認可コード → アクセストークン・リフレッシュトークン取得""" cred = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode() async with httpx.AsyncClient() as client: resp = await client.post( TOKEN_URL, headers={ "Content-Type": "application/x-www-form-urlencoded", "Authorization": f"Basic {cred}", }, data={ "grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri, "code_verifier": verifier, }, timeout=30, ) resp.raise_for_status() return resp.json() async def refresh_tokens_request(refresh_token: str, client_id: str, client_secret: str) -> dict: """リフレッシュトークンでアクセストークン再取得""" cred = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode() async with httpx.AsyncClient() as client: resp = await client.post( TOKEN_URL, headers={ "Content-Type": "application/x-www-form-urlencoded", "Authorization": f"Basic {cred}", }, data={ "grant_type": "refresh_token", "refresh_token": refresh_token, }, timeout=30, ) resp.raise_for_status() return resp.json() async def do_logout(refresh_token: str, client_id: str, client_secret: str): """ログアウト（トークン無効化）""" cred = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode() async with httpx.AsyncClient() as client: try: await client.post( LOGOUT_URL, headers={ "Content-Type": "application/x-www-form-urlencoded", "Authorization": f"Basic {cred}", }, data={"refresh_token": refresh_token}, timeout=15, ) except Exception: pass # ── 電子送達 API ────────────────────────────────────────────── async def get_post_list(access_token: str, date_from: str, date_to: str, limit: int = 50) -> dict: """ 電子送達一覧取得 GET /post/lists?date_from=YYYY-MM-DD&date_to=YYYY-MM-DD&limit=N&offset=1 """ async with httpx.AsyncClient() as client: resp = await client.get( f"{API_BASE}/post/lists", headers={"Authorization": f"Bearer {access_token}"}, params={"date_from": date_from, "date_to": date_to, "limit": limit, "offset": 1}, timeout=30, ) resp.raise_for_status() return resp.json() async def get_post_file(access_token: str, post_id: str) -> dict: """ 電子送達取得（通知文書ファイルを base64 で返す） GET /post/{post_id} Returns: {notice_data_name, file_data (base64), file_name_list} """ async with httpx.AsyncClient() as client: resp = await client.get( f"{API_BASE}/post/{post_id}", headers={"Authorization": f"Bearer {access_token}"}, timeout=60, ) resp.raise_for_status() return resp.json() async def mark_post_done(access_token: str, post_id: str) -> dict: """ 電子送達取得完了登録 POST /post {"post_id": "..."} """ async with httpx.AsyncClient() as client: resp = await client.post( f"{API_BASE}/post", headers={ "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", }, json={"post_id": post_id}, timeout=30, ) resp.raise_for_status() return resp.json()