AIでインシデント対応を民主化したい！日本語セキュリティLLM開発の挑戦と挫折、そして発見 #Security – Qiita

この記事が、AI×セキュリティというフロンティアで奮闘する方、特に日本語という言語の壁に悩む（私のような）エンジニアにとって、何かしらのヒントや勇気になれば幸いです。

私がこのツールを作ろうと思った原点は、社内SEとしてセキュリティを担当していた頃の経験にあります。以前の職場では、セキュリティ担当者はいたものの、専門知識を持つのはごく一部のメンバーに限られていました。他のメンバーは他業務と兼務しており、多忙の中で高度なセキュリティ知識をキャッチアップし続けるのは困難な状況でした。

結果として何が起こるか。インシデントは、こちらの都合などお構いなしに、深夜でも、休日でも発生します。そのたびに鳴り響くアラート。しかし、そのアラートが本当に危険なものなのか、的確な初動判断を下せるのは、結局限られたメンバーだけになってしまうのです。

• 「この海外からの不審なアクセス、ただの調査ポートスキャン？ それとも本格的な侵入の予兆？」
• 「このマルウェア検知アラート、既知のパターン？ それとも未知の脅威？」

こうした判断には、脅威インテリジェンスや過去の事例に関する深い知識が求められます。対応は特定の「スーパーマン」的なメンバーに集中し、チーム全体が疲弊していく…。この「専門知識の属人化」は、多くの組織が抱える根深い課題ではないでしょうか。

もし、そのスーパーマンの知識や判断プロセスをAIに移植できたら？ 専門家でないメンバーでも、AIに自然言語で相談することで、次の一手を導き出せるのではないか？

そんな思いから、「サイバー攻撃の状況を自然言語で入力すると、LLMが危険度やとるべきアクションを日本語で教えてくれる支援ツール」の開発はスタートしました。

「セキュリティAIなんて、MicrosoftやGoogleがすごいものを作っているじゃないか」

その通りです。開発に着手するにあたり、まずは既存の巨人たちが提供するサービスを徹底的に調査しました。Microsoft Security Copilot、Google Cloud Security AI Workbenchなど、素晴らしいサービスが既に存在し、進化を続けています。

しかし、調査を進める中で、「私たちの現場で本当に欲しい機能」との間に、いくつかのギャップがあることにも気づきました。

既存サービスとの比較分析

この調査から見えてきたのは、以下の点です。

1. 「予測」へのフォーカス:

多くのサービスが「現状分析」に強い一方、「次に何が起こるか」という攻撃予測まで踏み込んでいるものは少ない。

2.　日本語での対話能力:

日本語での自然言語入力や、非専門家向けの平易な解説は、まだ発展途上の領域。

3.　柔軟性:

特定の製品エコシステムに依存せず、任意のログファイルやテキスト情報を気軽に放り込めるような柔軟性は、意外と少ない。

つまり、「日本語で、対話的に、攻撃の”次の一手”を予測してくれる」という、私たちが現場で本当に欲しかった機能は、まだ誰も完成させていないブルーオーシャンだったのです。これが、巨人の肩を借りるだけでなく、自ら手を動かしてでも開発に挑戦する大きな動機となりました。

システム全体のアーキテクチャ構想

# （参考）DeepL APIで翻訳するコードの主要部分
import deepl
import json
import os
from dotenv import load_dotenv

load_dotenv()
auth_key = os.getenv("DEEPL_API_KEY") # .envファイルからAPIキーを読み込み
translator = deepl.Translator(auth_key)

def translate_text(text, target_lang="JA"):
    """DeepL APIを使ってテキストを翻訳する関数"""
    try:
        result = translator.translate_text(text, target_lang=target_lang)
        return result.text
    except Exception as e:
        print(f"Error during translation: {e}")
        return text # エラー時は原文を返す

def translate_stix_bundle(bundle):
    """STIXバンドル内の各オブジェクトを再帰的に翻訳"""
    for obj in bundle.get("objects", []):
        if "description" in obj:
            description_en = obj["description"]
            description_jp = translate_text(description_en)
            # STIXのカスタムプロパティとして翻訳結果を格納
            obj["x_mitre_description_jp"] = description_jp
            print(f"Translated: {obj.get('name', obj.get('id'))}")
    return bundle

# ...（ファイルの読み書き処理）...

Reconnaissance (偵察) -> [ATT&CK: TA0043, TA0042]
Weaponization (武器化) -> ...
Delivery (配送) -> [ATT&CK: TA0001]
...
Actions on Objectives (目的の実行) -> [ATT&CK: TA0002, TA0040, etc.]

# 指示プロンプトの例

あなたは経験豊富なSOCアナリストです。
以下のサイバー攻撃技術に関する情報を用いて、
セキュリティ初心者がするであろう質問(Instruction)と、
あなたが専門家として返すであろう詳細な回答(Response)の
ペアをJSON形式で生成してください。

# 入力情報
- 技術名: External Remote Services (T1133)
- 説明: 攻撃者は、VPNやCitrixなどの外部向けリモートサービスを悪用して、
        ネットワークへの初期アクセスや永続化を試みることがあります...
- キルチェーンフェーズ: Initial Access (初期アクセス)

# 出力形式
{
  "instruction": "（ここに初心者の質問を生成）",
  "response": "（ここに専門家の回答を生成）"
}

# （参考）FastAPIでRAG+AgentのストリーミングAPIを実装する構想
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
# ... LangChainやLLMモデルのインポート ...

app = FastAPI()

# 事前に日本語知識ベースを読み込んでおく
# retriever = ... (knowledge_base.jsonからVectorStoreRetrieverを作成)
# llm = ... (SFT済み、またはベースのLLMをロード)

class Query(BaseModel):
    text: str

async def agent_stream(query: str):
    # LangChainのAgent Executorを使ってストリーミング応答を生成
    # agent_executor = create_agent(llm, retriever)
    async for chunk in agent_executor.astream({"input": query}):
        # ... chunkを処理してyield ...
        yield f"data: {json.dumps(chunk)}\n\n"

@app.post("/chat")
async def chat_endpoint(query: Query):
    return StreamingResponse(agent_stream(query.text), media_type="text/event-stream")