MENU

問い合わせ


    【Claude CodeとAWSで作るAIエージェントの土台 第7回】既存のリポジトリから移す:秘密の値の検査と、WordPress への投稿の中継

    前回は、CLAUDE.md を入口だけにし、役割ごとの指示書を API から配る形を作りました。

    今回は、今まで使っていたリポジトリから土台へ移す話と、この連載の目的だった「WordPress への投稿」を API に足す話です。

    「今のリポジトリをそのまま新しい仕組みに移していいの? 何か確かめておくことは?」

    結論から言うと、移す前に秘密の値を機械的に検査し、見つかった値は「置き場を移す」だけでなく「作り直す」 ことが大切です。git の履歴に残った値は、ファイルから消しても取り出せるからです。そして WordPress のパスワードは、AI が触れない Secrets Manager に置き、API が投稿のときだけ付けます。

    目次

    移す前に、秘密の値を検査する

    移す前に、全ファイルを機械的に検査しました。当社の場合は探したのは次の形です。

    • Google の APIキー
    • WordPress のアプリケーションパスワード
    • 秘密鍵
    • AWS・GitHub・AI サービスのキーの形

    もしこれらの値がファイルに記載されていた場合、ファイルから消しても、git の履歴には残ります。リポジトリを見られる人なら、過去の版から値を取り出せます。そのため、見つかった値は置き場を移すだけでなく、作り直して古い値を使えなくしました。

    誤検知もあった

    機械的な検査には、誤検知(=秘密ではないのに秘密と判定されること)もありました。

    • 記事に出てくる社名の一部が、ある AI サービスのキーの形(sk- で始まる並び)と同じだった
    • 記事に埋め込む前提の地図の APIキーが見つかった。これは公開される前提のキーなので秘密ではありません。代わりに「使えるサイトの制限がかかっているか」を確かめました

    誤検知を見て「検査は当てにならない」と止めるのではなく、1件ずつ見て、秘密かどうかを人が判断します。

    値の置き場の決め方

    見つかった値や、これから使う値は、使う場面ごとに置き場を決めました。

    値置き場
    WordPress・Google のキー(セッションから使う)Secrets Manager。API が中継のときに付ける
    毎朝の価格更新などの自動処理のキーGitHub Actions の Secrets のまま
    AI の APIキー(文章の要約に使っていた)不要にした。セッションの Claude に要約させる

    AI の APIキーは、スクリプトを「依頼文を書き出す」「答えを差し込む」の2段に分け、間の要約をセッションの Claude に任せることで、キーそのものを不要にしました。持たなくてよいキーは、持たないのがいちばん安全です。

    確認用と本番では、値も分けます。WordPress のアプリケーションパスワードも Google の APIキーも、環境ごとに作りました。

    WordPress への投稿を中継する

    ここからはコードです。WordPress のアプリケーションパスワード(=ログイン用のパスワードとは別に、プログラムからの利用のために発行するパスワード)は、AWS の Secrets Manager に置き、API が投稿のときだけ読みます。

    📰 出典:Application Passwords(WordPress Developer Resources)

    値を Secrets Manager に入れる

    WordPress の管理画面の「ユーザー」→「プロフィール」で、アプリケーションパスワードを発行します。投稿の権限があるユーザーで、確認用と本番で別々に発行します。

    発行した値は、Secrets Manager の画面で「Store a new secret」から「Other type of secret」を選び、次の2つのキーで保存します(画面の言語:英語。日本語表示では名前が異なる場合があります)。

    キー値
    usernameWordPress のユーザー名
    app_password発行したアプリケーションパスワード

    名前は ai-agent-kiban/staging/wp/sample(本番は ai-agent-kiban/prod/wp/sample)にします。第4回で、API には ai-agent-kiban/<stage>/ で始まる値だけを読む権限を付けたので、この名前に合わせる必要があります。値をコマンドの引数に書くと、ターミナルの履歴に残るので、画面から入れるのが無難です。

    WordPress を呼ぶ部分(app/wordpress.py)

    # app/wordpress.py
    """WordPress への投稿(中継)。
    
    WordPress のユーザー名とアプリケーションパスワードは、AWS の Secrets Manager の
    `<SECRET_PREFIX>/wp/<サイト>` に {"username": ..., "app_password": ...} で置く。
    API の呼び出し元(Claude Code のセッション)には返さず、ここで使うだけにする。
    """
    
    import base64
    import json
    
    
    class Secrets:
        def __init__(self, prefix: str, client=None, values: dict | None = None):
            self._prefix = prefix
            self._client = client
            self._values = values  # テスト用(Secrets Manager の代わり)
            self._cache: dict = {}
    
        def get(self, name: str) -> dict:
            if self._values is not None:
                return self._values[name]
            if name not in self._cache:
                if self._client is None:
                    import boto3
                    self._client = boto3.client('secretsmanager')
                res = self._client.get_secret_value(SecretId=f'{self._prefix}/{name}')
                self._cache[name] = json.loads(res['SecretString'])
            return self._cache[name]
    
    
    async def create_post(http, base_url: str, secret: dict, title: str, content: str, status: str) -> dict:
        basic = base64.b64encode(f"{secret['username']}:{secret['app_password']}".encode()).decode()
        res = await http.post(
            base_url.rstrip('/') + '/wp-json/wp/v2/posts',
            json={'title': title, 'content': content, 'status': status},
            headers={'Authorization': f'Basic {basic}'},
            timeout=30,
        )
        res.raise_for_status()
        body = res.json()
        return {'id': body['id'], 'link': body.get('link', ''), 'status': body.get('status', status)}
    • Secrets は Secrets Manager から値を読み、一度読んだ値は覚えておきます。テストでは values に偽物の値を渡します
    • create_post は WordPress の REST API(=プログラムから WordPress を操作するための入口)に投稿します。パスワードは Basic 認証のヘッダーとして付けるだけで、戻り値には含めません

    API に投稿の入口を足す

    app/main.py に次の変更を加えます(+ が追加、- が削除)。

    --- a/app/main.py
    +++ b/app/main.py
    @@ -3,6 +3,7 @@
       GET  /healthz                      起動確認(トークン不要)
       GET  /me                           自分の役割
       GET  /instructions                 共通+自分の役割の指示書(セッションの最初に読む)
    +  POST /sites/{site}/posts           {"title": ..., "content": ..., "status": "draft"|"publish"}
    
     起動:uvicorn --factory app.main:create_app
     """
    @@ -10,20 +11,32 @@
     import json
     from pathlib import Path
    
    -from fastapi import Depends, FastAPI
    +import httpx
    +from fastapi import Depends, FastAPI, HTTPException
     from fastapi.responses import PlainTextResponse
    +from pydantic import BaseModel, Field
    
     from .auth import Principal, TokenStore, get_principal
    +from .roles import can_post
     from .settings import Settings
    +from .wordpress import Secrets, create_post
    
    
    -def create_app(settings: Settings | None = None, *, tokens=None) -> FastAPI:
    +class PostIn(BaseModel):
    +    title: str = Field(min_length=1, max_length=200)
    +    content: str = Field(min_length=1)
    +    status: str = Field(default='draft', pattern='^(draft|publish)$')
    +
    +
    +def create_app(settings: Settings | None = None, *, tokens=None, secrets=None, http=None) -> FastAPI:
         settings = settings or Settings()
         app = FastAPI(title='ai-agent-kiban')
         app.state.settings = settings
         app.state.sites = {k: v for k, v in json.loads(Path(settings.sites_file).read_text(encoding='utf-8')).items()
                            if not k.startswith('_')}
         app.state.tokens = tokens or TokenStore(settings.tokens_table, settings.local_tokens_file)
    +    app.state.secrets = secrets or Secrets(settings.secret_prefix)
    +    app.state.http = http or httpx.AsyncClient(follow_redirects=False)
    
         @app.get('/healthz')
         def healthz():
    @@ -41,4 +54,18 @@
             return '\n\n'.join(f'<!-- {f.relative_to(base)} -->\n' + f.read_text(encoding='utf-8') for f in files) \
                 or f'({p.role} の指示書はまだありません)'
    
    +    @app.post('/sites/{site}/posts')
    +    async def post(site: str, body: PostIn, p: Principal = Depends(get_principal)):
    +        if site not in app.state.sites:
    +            raise HTTPException(404, f'サイト {site} はありません')
    +        if not can_post(p.role, site):
    +            raise HTTPException(403, f'役割 {p.role} は {site} に投稿できません')
    +        status = body.status if settings.can_publish else 'draft'  # 確認用は必ず下書き
    +        try:
    +            result = await create_post(app.state.http, app.state.sites[site]['base_url'],
    +                                       app.state.secrets.get(f'wp/{site}'), body.title, body.content, status)
    +        except httpx.HTTPStatusError as e:
    +            raise HTTPException(502, f'WordPress がエラーを返しました({e.response.status_code})') from e
    +        return result | {'requested_status': body.status}
    +
         return app

    投稿の入口では、3つの順で確かめます。

    1. サイトが config/sites.json にあるか(なければ 404)
    2. その役割が投稿できるサイトか(第1回の can_post。できなければ 403)
    3. 本番か(確認用なら、公開を頼まれても下書きにする)

    WordPress がエラーを返したときは 502 にし、WordPress の返した中身は呼び出し元にそのまま返しません。中身に設定の情報が含まれていることがあるためです。

    PostIn の Field で、タイトルの長さや status の値(draft か publish だけ)を確かめています。形の正しくない頼み方は、WordPress に届く前に API が断ります。

    テストで確かめる

    テストでは、WordPress の代わりに、受け取った呼び出しを覚えておく偽物を使います。tests/conftest.py に次の変更を加えます。

    --- a/tests/conftest.py
    +++ b/tests/conftest.py
    @@ -1,9 +1,13 @@
    +import json
    +
    +import httpx
     import pytest
     from fastapi.testclient import TestClient
    
     from app.auth import TokenStore, hash_token
     from app.main import create_app
     from app.settings import Settings
    +from app.wordpress import Secrets
    
     TOKENS = {
         'th_ok': {'role': 'blog-sample'},
    @@ -24,14 +28,40 @@
             return self._items.get(token_hash)
    
    
    -def make(stage):
    +class FakeWordPress:
    +    """WordPress の代わり。受け取った呼び出しを覚えておく"""
    +
    +    def __init__(self):
    +        self.calls = []
    +
    +    def handler(self, request: httpx.Request):
    +        self.calls.append(request)
    +        body = json.loads(request.content)
    +        return httpx.Response(201, json={'id': 123, 'link': 'https://blog.example.com/?p=123',
    +                                         'status': body['status']})
    +
    +
    +def make(stage, wp):
         settings = Settings(stage=stage, sites_file='config/sites.json')
    -    return TestClient(create_app(settings, tokens=FakeTokens()))
    +    secrets = Secrets('x', values={'wp/sample': {'username': 'writer', 'app_password': 'xxxx'}})
    +    app = create_app(settings, tokens=FakeTokens(), secrets=secrets,
    +                     http=httpx.AsyncClient(transport=httpx.MockTransport(wp.handler)))
    +    return TestClient(app)
    
    
     @pytest.fixture
    -def prod():
    -    return make('prod')
    +def wp():
    +    return FakeWordPress()
    +
    +
    +@pytest.fixture
    +def prod(wp):
    +    return make('prod', wp)
    +
    +
    +@pytest.fixture
    +def staging(wp):
    +    return make('staging', wp)
    
    
     def bearer(token):

    tests/test_api.py に次の変更を加えます。

    --- a/tests/test_api.py
    +++ b/tests/test_api.py
    @@ -1,3 +1,5 @@
    +import json
    +
     from conftest import bearer
    
    
    @@ -10,3 +12,27 @@
         for t in ('nope', 'th_revoked', 'th_expired'):
             assert prod.get('/me', headers=bearer(t)).status_code == 401, t
         assert prod.get('/me', headers=bearer('th_ok')).json() == {'role': 'blog-sample'}
    +
    +
    +def test_post_goes_to_wordpress_with_the_key_added_by_the_api(prod, wp):
    +    res = prod.post('/sites/sample/posts', headers=bearer('th_ok'),
    +                    json={'title': 'テスト', 'content': '<p>本文</p>', 'status': 'publish'})
    +    assert res.status_code == 200 and res.json()['id'] == 123 and res.json()['status'] == 'publish'
    +    req = wp.calls[-1]
    +    assert str(req.url) == 'https://blog.example.com/wp-json/wp/v2/posts'
    +    assert req.headers['authorization'].startswith('Basic ')  # パスワードは API が付ける
    +    assert 'xxxx' not in res.text  # 呼び出し元には返さない
    +
    +
    +def test_role_can_only_post_to_its_own_sites(prod, wp):
    +    res = prod.post('/sites/sample/posts', headers=bearer('th_other'), json={'title': 't', 'content': 'c'})
    +    assert res.status_code == 403 and not wp.calls
    +    assert prod.post('/sites/nowhere/posts', headers=bearer('th_ok'),
    +                       json={'title': 't', 'content': 'c'}).status_code == 404
    +
    +
    +def test_staging_only_makes_drafts(staging, wp):
    +    res = staging.post('/sites/sample/posts', headers=bearer('th_ok'),
    +                       json={'title': 't', 'content': 'c', 'status': 'publish'})
    +    assert res.json()['status'] == 'draft' and res.json()['requested_status'] == 'publish'
    +    assert json.loads(wp.calls[-1].content)['status'] == 'draft'
    • パスワードは API が付けていて、呼び出し元への返事には含まれない
    • ほかの役割のトークンでは 403 になり、WordPress には何も届かない
    • 確認用では、公開を頼んでも下書きになる

    セッションから投稿を頼むコマンド

    tools/post.py(1行目は実行用の指定なので、パスは書いていません)

    #!/usr/bin/env python3
    """Claude Code のセッションから API に投稿を頼むコマンド。
    
      python3 tools/post.py --site sample --title "タイトル" --file article.html [--publish]
    
    接続先は環境変数 TH_API_URL(例: https://api.example.com)。
    トークンは Claude Code の環境の Network secrets が自動で付けるので、このコマンドには渡さない。
    手元で試すときだけ、環境変数 TH_TOKEN を使う。
    """
    
    import argparse
    import json
    import os
    import sys
    import urllib.error
    import urllib.request
    from pathlib import Path
    
    
    def main():
        ap = argparse.ArgumentParser()
        ap.add_argument('--site', required=True)
        ap.add_argument('--title', required=True)
        ap.add_argument('--file', required=True, help='本文(HTML かテキスト)')
        ap.add_argument('--publish', action='store_true', help='公開する(付けなければ下書き)')
        a = ap.parse_args()
    
        base = os.environ.get('TH_API_URL', '').rstrip('/')
        if not base:
            sys.exit('TH_API_URL が設定されていません')
        body = json.dumps({'title': a.title, 'content': Path(a.file).read_text(encoding='utf-8'),
                           'status': 'publish' if a.publish else 'draft'}).encode('utf-8')
        req = urllib.request.Request(f'{base}/sites/{a.site}/posts', data=body, method='POST',
                                     headers={'Content-Type': 'application/json'})
        if os.environ.get('TH_TOKEN'):
            req.add_header('Authorization', f"Bearer {os.environ['TH_TOKEN']}")
        try:
            with urllib.request.urlopen(req, timeout=60) as res:
                print(json.dumps(json.load(res), ensure_ascii=False, indent=2))
        except urllib.error.HTTPError as e:
            sys.exit(f'{e.code}: {e.read().decode("utf-8", "replace")}')
    
    
    if __name__ == '__main__':
        main()

    tools/agent.py の説明にも、投稿のコマンドを書き足します。

    --- a/tools/agent.py
    +++ b/tools/agent.py
    @@ -5,7 +5,7 @@
       python3 tools/agent.py instructions    共通+自分の役割の指示書(セッションの最初に読む)
    
     接続先は環境変数 TH_API_URL。トークンは Claude Code の環境の Network secrets が付ける。
    -手元で試すときだけ、環境変数 TH_TOKEN を使う。
    +手元で試すときだけ、環境変数 TH_TOKEN を使う。投稿は tools/post.py。
     """
    
     import argparse

    役割の指示書にも、投稿のしかたを書きます。instructions/blog-sample/10_blog-sample.md に次の変更を加えます。

    --- a/instructions/blog-sample/10_blog-sample.md
    +++ b/instructions/blog-sample/10_blog-sample.md
    @@ -4,7 +4,8 @@
    
     ## できること
    
    -- サイト `sample` に載せる記事の本文を書き、ファイルに保存する
    +- サイト `sample` への投稿:`python3 tools/post.py --site sample --title "<タイトル>" --file <本文のファイル>`
    +  - 公開するのは、使う人に「公開して」と頼まれたときだけ `--publish` を付ける
    
     ## できないこと
    

    「公開するのは、頼まれたときだけ」は指示書のお願いです。確認用では API が必ず下書きにするので、守りは API の側にもあります。

    移行で起きたこと

    資料の正を1か所にする

    移行のあとも、移行元(git のリポジトリ)で記事を作る仕組みが動いていて、困りごとの台帳が2か所で書き換えられていました。切り替えた役割から順に、移行元での作業を止めました。

    また、移行の道具をやり直すと、セッションが S3 で書き換えた台帳や資料を、古い内容で上書きしてしまうことに気付きました。S3 に既にあって中身が違うものは送らずに一覧に出し、上書きは明示したときだけにしました。移行の道具は人が CloudShell で実行し、AI には「何が送られるかの一覧を出す」ところまでしかさせていません。

    AI にリポジトリを丸ごと取り込ませない

    役割の指示書がスクリプトを案内していたのに、そのスクリプトが古い一体型のリポジトリにしか無く、営業のセッションの AI がそのリポジトリを丸ごと取り込んで使おうとしたことがありました。Claude Code の安全チェックが「外から取ってきたコードの実行」として止めたことで気付きました。

    対処として、共通のスクリプトは土台のリポジトリ(API と同じ場所)に移し、資料は API から取るようにしました。前回の「指示書が案内するスクリプトがあるか」のテストは、このときに足したものです。

    動作確認

    pip install -e ".[dev]"
    ruff check .
    pytest -q

    テストが13件通れば完成です。WordPress への実際の投稿まで試すには、AWS の Secrets Manager に値が入っていて、API がそれを読める必要があります(AWS に反映した API で試します)。セッションからは次のように頼みます。

    python3 tools/post.py --site sample --title "テストの投稿" --file article.html

    確認用の API では、--publish を付けても "status": "draft"(下書き)で返ります。

    つまずきやすい点・セキュリティ上の注意

    • 見つかった秘密の値は作り直す:ファイルから消すだけでは、git の履歴から取り出せます
    • WordPress のエラーの中身をそのまま返さない:設定の情報が含まれることがあります
    • アプリケーションパスワードは、必要な権限のユーザーで発行する:管理者のユーザーで発行すると、投稿以外の操作もできてしまいます。投稿者(や編集者)の権限のユーザーを用意すると安全です
    • 確認用の API は外に書き込まない:サンプルでは下書きにしていますが、確認用のサイトを別に用意できるなら、そちらに向けるとさらに安全です

    発注者向けメモ

    今あるリポジトリやツールを新しい仕組みに移すときは、次を開発会社に確認しましょう。

    • 「移す前に、パスワードや APIキーが入っていないかを検査しますか。見つかったら作り直しますか」
    • 「移行のあと、古い仕組みはいつ止めますか。両方で同じデータを書き換える期間はありますか」
    • 「外部サービスのパスワードは、どこに置きますか。AI やツールの利用者から見えますか」
    • 「確認用の環境から、本番のサービス(WordPress など)に書き込むことはありますか」

    移すファイルが多い、秘密の値が多く見つかる、古い仕組みを並行して動かす期間が長い、といった条件があると、工数が増えやすくなります。

    まとめと次回予告

    • 移す前に秘密の値を機械的に検査し、見つかった値は作り直す。誤検知は1件ずつ人が判断する
    • 値の置き場は使う場面で決める。持たなくてよいキーは持たない
    • WordPress のパスワードは Secrets Manager に置き、API が投稿のときだけ付ける。確認用では必ず下書きにする

    次回は最終回です。401 の記録と監視、WAF での回数の制限、確認用をふだん止めて費用を抑える方法 を作り、連載をまとめます。

    この連載の記事一覧

    この記事は連載「Claude CodeとAWSで作るAIエージェントの土台」の1回です。連載のほかの回は次のとおりです(連載の一覧ページ)。

    システム制作・運用・保守のお問い合わせはこちら


      よかったらシェアしてね!
      • URLをコピーしました!
      • URLをコピーしました!

      この記事を書いた人

      コメント

      目次