LINEミニアプリと連携！Flaskで外部向けAPIを作ってNginxリバースプロキシで安全に公開する

smarthome/
└── interfaces/
    └── inbound/
        ├── http_server.py          # Flaskアプリ本体（エントリーポイント）
        ├── api/                    # 外部向けAPI Blueprint
        │   ├── __init__.py         # Blueprint定義
        │   ├── dispatcher.py       # 認証デコレーター・APIエンドポイント
        │   └── routes.py           # シンプルなルート（pingなど）
        ├── internal/               # LAN内向けAPI Blueprint
        │   ├── __init__.py
        │   └── dispatcher.py
        └── webhooks/               # Webhook受信 Blueprint
            ├── __init__.py
            └── routes.py

# プロジェクトフォルダに移動
cd ~/Projects

# 仮想環境を作成
python3 -m venv venv

# 仮想環境を有効化
source venv/bin/activate

pip install flask flask-limiter requests

pip show flask flask-limiter
# Name: Flask
# Version: 3.1.2
# Name: Flask-Limiter
# Version: 4.1.1

from flask import Blueprint

# 外部向けAPIをまとめるBlueprint
api_bp = Blueprint(
    "api",          # Blueprint識別名（アプリ内で一意）
    __name__,       # このBlueprintが属するモジュール名
    url_prefix="/api"
)

# dispatcherとroutesを読み込むことでエンドポイントをBlueprintに登録する
from . import dispatcher
from . import routes

__all__ = ["api_bp"]

from functools import wraps
from flask import g, request, abort, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
import time
import requests

from . import api_bp

# ----------------------------------------
# Flask-Limiter の初期設定
# ----------------------------------------
limiter = Limiter(
    key_func=get_remote_address,
    default_limits=[]   # デフォルトは制限なし（エンドポイントごとに個別設定）
)

# ----------------------------------------
# 設定
# ----------------------------------------
LINE_CHANNEL_ID = "xxxxxxxxxx"   # LINE DevelopersのChannel ID

# 許可するLINEユーザーのIDリスト（各ユーザーのsub値）
LINE_USER_MAP = {
    "Uxxxxxxxxxxxxxxxxxxxx": {"name": "user1"},
    "Uyyyyyyyyyyyyyyyyyyyyyy": {"name": "user2"},
}

# ----------------------------------------
# LINE IDトークン検証（LINE APIに問い合わせ）
# ----------------------------------------
def verify_line_id_token(id_token):
    url = "https://api.line.me/oauth2/v2.1/verify"
    data = {
        "id_token": id_token,
        "client_id": LINE_CHANNEL_ID
    }
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    r = requests.post(url, data=data, headers=headers, timeout=3)

    if r.status_code != 200:
        return None
    return r.json()

# ----------------------------------------
# IDトークン検証キャッシュ（有効期限内は再検証しない）
# ----------------------------------------
token_cache = {}

def verify_line_id_token_cached(id_token):
    now = time.time()
    # 期限切れキャッシュの削除
    for token in list(token_cache.keys()):
        if token_cache[token]["exp"] <= now:
            del token_cache[token]
    # キャッシュヒットなら返す
    if id_token in token_cache:
        cached = token_cache[id_token]
        if cached["exp"] > now:
            return cached["data"]
    # キャッシュミス：LINE APIで検証してキャッシュに保存
    user_info = verify_line_id_token(id_token)
    if user_info:
        token_cache[id_token] = {
            "data": user_info,
            "exp": user_info.get("exp", 0)
        }
    return user_info

# ----------------------------------------
# 認証デコレーター
# ----------------------------------------
def require_line_auth(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        auth = request.headers.get("Authorization", "")
        if not auth.startswith("Bearer "):
            return jsonify({"status": "error", "reason": "missing token"}), 401

        id_token = auth.replace("Bearer ", "")
        user_info = verify_line_id_token_cached(id_token)

        if not user_info:
            return jsonify({"status": "error", "reason": "invalid token"}), 401

        user_id = user_info.get("sub")
        if user_id not in LINE_USER_MAP:
            return jsonify({"status": "error", "reason": "forbidden"}), 403

        # エンドポイント関数内で g.user からユーザー情報を参照できる
        g.user = {
            "line": user_info,
            "config": LINE_USER_MAP[user_id]
        }
        return f(*args, **kwargs)
    return wrapper

# ----------------------------------------
# APIエンドポイント
# ----------------------------------------

@api_bp.route('/api/check_user', methods=['GET'])
@require_line_auth
def check_user():
    # 認証テスト用エンドポイント
    return jsonify({"status": "ok"}), 200


@api_bp.route('/api/speak_googlehome', methods=['POST'])
@require_line_auth
@limiter.limit("10 per minute")   # 1分間に10回まで
def speak_googlehome():
    # Google Homeに指定メッセージを発話させる
    try:
        data = request.get_json()
        message = data.get("message")

        if not message or len(message) > 255:
            return jsonify({"status": "error", "reason": "invalid message"}), 400

        # Google Homeへの発話処理（別モジュールで実装）
        # google_notify.speak(message)

        return jsonify({"status": "ok"}), 200

    except Exception:
        return jsonify({"status": "error"}), 500

from flask import Flask

from interfaces.inbound.api import api_bp
from interfaces.inbound.api.dispatcher import limiter
from interfaces.inbound.internal import internal_bp
from interfaces.inbound.webhooks import webhooks_bp

app = Flask(__name__)
limiter.init_app(app)   # Flask-LimiterをFlaskアプリに紐付ける

# Blueprintを登録（url_prefixで各Blueprintのベースパスを設定）
app.register_blueprint(api_bp,       url_prefix="/api")
app.register_blueprint(internal_bp,  url_prefix="/internal")
app.register_blueprint(webhooks_bp,  url_prefix="/webhooks")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

sudo nano /etc/systemd/system/smh-http-server.service

[Unit]
Description=Smarthome HTTP Server (Flask)
After=nginx.service
Wants=nginx.service

[Service]
# 仮想環境のPythonでhttp_server.pyを起動する
ExecStart=/home/user/Projects/venv/bin/python /home/user/Projects/smarthome/interfaces/inbound/http_server.py
Restart=always          # 異常終了時に自動再起動
RestartSec=60           # 再起動までの待機時間（秒）
Type=simple
User=user               # 実行ユーザー（rootは使わない）

[Install]
WantedBy=multi-user.target

# systemdにサービスファイルを認識させる
sudo systemctl daemon-reload

# OS起動時に自動起動する設定 & 今すぐ起動
sudo systemctl enable --now smh-http-server

# 起動状態を確認
sudo systemctl status smh-http-server

server {
    listen 443 ssl;
    server_name xxxx.tplinkdns.com;

    # SSL設定（省略）...

    # -----------------------------------------
    # /external/ : 外部向けFlask API
    # -----------------------------------------
    location /external/ {
        proxy_pass http://127.0.0.1:8000/api/;          # /external/ → /api/ にパス変換
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_cache_bypass $http_upgrade;

        # LINEのBearerトークンをFlaskに転送する（これがないと認証が通らない）
        proxy_set_header Authorization $http_authorization;

        # アクセスログを外部API用ファイルに分離
        access_log /srv/nginx/log/api.log detailed;
    }

    # -----------------------------------------
    # /webhooks/ : 外部Webhook受信
    # -----------------------------------------
    location /webhooks/ {
        proxy_pass http://127.0.0.1:8000/webhooks/;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_cache_bypass $http_upgrade;
    }
}

server {
    listen 80;
    server_name 192.168.68.89;

    # -----------------------------------------
    # /internal/ : LAN内専用API（IPアドレス制限あり）
    # -----------------------------------------
    location /internal/ {
        proxy_pass http://127.0.0.1:8000/internal/;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        allow 192.168.68.0/24;   # LAN内のみ許可
        deny all;

        access_log /srv/nginx/log/api_internal.log detailed;
    }
}

# 構文チェック
sudo nginx -t

# 問題なければリロード
sudo systemctl reload nginx

@api_bp.route('/api/speak_googlehome', methods=['POST'])
@require_line_auth
@limiter.limit("10 per minute")   # 同じIPから1分間に10回まで
def speak_googlehome():
    ...

sudo systemctl status smh-http-server
# Active: active (running) と表示されればOK

# ポート8000でListenしているか確認
ss -tlnp | grep 8000
# LISTEN  0  128  0.0.0.0:8000  ...

# サーバー内からFlaskに直接アクセス（認証なし）
curl http://127.0.0.1:8000/api/ping
# {"status": "ok"}

# Nginx経由でアクセス（外部URL）
curl https://xxxx.tplinkdns.com/external/api/ping
# {"status": "ok"}

# 認証なしでアクセス → 401 Unauthorized
curl https://xxxx.tplinkdns.com/external/api/check_user
# {"reason":"missing token","status":"error"}

# 有効なLINE IDトークンを付けてアクセス → 200 OK
curl -H "Authorization: Bearer <LINEのIDトークン>" \
  https://xxxx.tplinkdns.com/external/api/check_user
# {"status":"ok"}