はじめに 2026年10月21日、PHPに存在する2件の重大な脆弱性が公表されました。組み合わせて「PHP2Shell」と呼ばれる問題で、条件がそろうと、外部からサーバー上で不正なコードを実行される恐れがあります。情報処理推進機構（IPA）も翌22日に注意喚起を公開しました。その後も追加の発表がなされているようですが、今回は、PHPとはどのようなもので、どういった仕組みに使われているのかを中心に説明いたします。 1. PHPとは ～ Webサイトが表示されるまでの仕組み PHPという名前を聞いたことがあっても、実際にどのような処理をしているのか、よく知らない方もいるかもしれません

1. はじめに 本記事では、Oracle WebLogic Server（以下、WebLogic）からApache Tomcat（以下、Tomcat）への移行検証を紹介します。 「次のシステム更改で、WebLogicをTomcatに移行するから。よろしく。」 と上司から言われて、さて困りました。 とはいえ、最初からものすごく難しく考えていたわけではありません。 Oracle WebLogic Server Proxy Plug-In（以下、WebLogic Proxy Plug-In）をApplication Request Routing（以下、ARR）に置き換えて、Tomcatま

はじめに 本記事は昨今増加しているボットアクセスの可視化をするために必要な、 データ連携フロー構築に関するナレッジになります。 関連するクラウドサービスはAWSおよびGoogleになります。 担当しているサービスがAWS基盤であり、集計および可視化がGoogle基盤のため データを加工して連携するロジックの共有 となります。 ※Google側に共有したあとの可視化までの流れは本記事のスコープ外とします。 同様の状況にあり、これから 可視化するためのフローを構築する方 に役立てば幸いです。 また、単にAWSとGCPのクロスクラウドデータ連携の必要がある方にも役立つかもしれません。 背景と状況 ・Webサーバーのアクセスログを活用してアクセス状況を可視化したい ・サービス(元データ側)はS3にWebサーバー(Apache)のログを日次で退避している ※サーバーからS3へ退避されている前提として、その辺は割愛させていただきます ・Google側には既に可視化基盤が整っている(BigQuery+Looker Studio) 全体像 今回のデータフロー全体像を先に示します。 利用サービス ・AWS S3 ・AWS Glue ・Google Storage Transfer Service ・Google Cloud Storage ・Google BigQuery ・Google Looker Studio 課題と解決策 データ加工 元データのApacheアクセスログと既存の可視化基盤の項目には差異があるため AWS側で加工してすぐ可視化できる状態で引き渡す必要がありました。 →データ加工はGlueジョブのスクリプト内で実施しました。 ボット種類による分類 Google系ボットのアクセスか、それ以外のAIボットのアクセスか によって分類して集計したい要望がありました。 とはいえApacheアクセスログには混在しているため、それらを切り分けて抽出する必要がありました。 →切り分けはGlueジョブ実行時のパラメータ(Glueトリガー側)で実施しました。 クロスクラウドデータ連携 BigQuery Data Transferが利用できず、Storage Transfer Serviceを初採用した点。 既存でナレッジのあった前者Data TransferではIAM ユーザーのアクセスキー、シークレットキーを利用するため セキュリティ懸念があり、今回は採用しない方針としました。 →後者Storage Transfer ServiceではAssumeRoleを利用するためセキュリティリスクが下がります。 　既存ナレッジもフロー検討時に参考にさせていただきましたが、 　時代に合わせて実装も変えていく必要があるということですね。 実装 Glue ジョブスクリプト 既存のGlueジョブスクリプトでデータ連携系のものもありましたが、 今回S3格納済ログファイルのデータ加工が発生するため完全新規で作成しました。 スクリプトはほとんどAIで生成したため、参考までにプロンプトも添えておきます。 AIプロンプト 文章は重要ではないのでキャプチャで割愛します。 (BigQuery Data TransferはStorage Transfer Serviceに読み替えてください) ※実装当初は東京リージョン未対応だったようですが、2026年9月現在は対応されているようです。 https://docs.cloud.google.com/storage-transfer/docs/source-amazon-s3?hl=ja ...もろもろ壁打ちの結果、できた設計書が以下です。 簡易設計書 AIとのやり取り、最終結果どうなのかを簡単にまとめる習慣が個人的にあるのですが、 スクリプト生成の際などはアウトプットとして設計書を出しておくのが有意義だと思います。 # botアクセス可視化向け データ加工設計書（AWS Glue）## 1. 概要本設計書は、Apacheアクセスログを対象に、botアクセスのみを抽出・加工し、 BIツール（BigQuery / Looker Studio）での可視化に適した形式で連携するための データ加工処理（AWS Glue）の設計方針を示すものである。---## 2. 全体構成### 2.1 処理フローCloudFront / EC2 Apache Log↓S3（ap-northeast-1）↓AWS Glue（バッチ・日次）↓S3（us-east-1）↓Storage Transfer Service↓Cloud Storage↓BigQuery↓Looker Studio### 2.2 担当範囲- 本設計書の対象範囲 - S3（ログ）→ AWS Glue → S3（加工済データ）- BIツール側（BigQuery / Looker Studio）はサイト運営チームの担当とする---## 3. インプット仕様### 3.1 入力データ格納先- リージョン：ap-northeast-1- バケット：s3://xxxxxxxxxxxxxxx/ec2-log/└ prod-webap0X/apache.access/{yyyy}/{MM}/{dd}/{hh}/### 3.2 ファイル形式- テキスト（TSV）- 列構成：| 列番号 | 内容 ||------|------|| 1 | 収集時刻（ISO8601） || 2 | サーバ名 / ログ種別 || 3 | Apacheアクセスログ（JSON文字列） |---## 4. データ加工概要### 4.1 処理単位- 日次バッチ- 前日分（yyyy/MM/dd）のみを対象に処理を行う### 4.2 処理内容1. Apacheログ(JSON)の構造化2. user-agent による botアクセス判定3. botアクセスのみ抽出4. BI向けにフィールドを正規化5. CSV形式で S3（us-east-1）へ出力---## 5. bot判定仕様### 5.1 判定対象フィールド- JSON内 `agent`（User-Agent）### 5.2 判定方法- user_agent を小文字化した上で、 判定用トークンとの **部分一致または完全一致** により判定する- いずれかのトークンに一致した場合、botアクセスとして採用する### 5.3 判定用トークン一覧OAI-SearchBotChatGPT-UserGPTBotClaude-SearchBotClaude-UserClaudeBotClaude-Webanthropic-aiGoogle-CloudVertexBotGoogle-ExtendedPerplexityBotPerplexity-UserApplebot-Extendedmeta-externalagentMeta-WebIndexerSBIntuitionsBotMistralAI-Usercohere-aicohere-training-data-crawlerimg2datasetChatGLM-SpiderAI2Botsider.ai---## 6. アウトプット仕様### 6.1 出力先- リージョン：us-east-1- バケット：s3://xxxxxxxxxxxxxxx/group-a/{yyyy}/{MM}/{dd}/### 6.2 ファイル形式- CSV（ヘッダあり）---## 7. 出力項目定義出力する項目は以下に限定し、 **項目名・順序ともに固定**とする。| No | 項目名 | マッピング元 | 備考 ||---|---|---|---|| 1 | date | 収集時刻 | yyyy-MM-dd || 2 | time | 収集時刻 | HH:mm:ss || 3 | x-edge-location | - | NULL固定 || 4 | c-ip | host | 先頭IPのみ || 5 | cs-method | method | || 6 | cs(Host) | - | NULL固定 || 7 | cs-uri-stem | path | || 8 | sc-status | code | 数値 || 9 | cs(Referer) | referer | "-"はNULL ||10 | cs(User-Agent) | agent | ||11 | time-taken | - | NULL固定 |---## 8. 非機能要件### 8.1 冪等性- 同一日付の再実行時は出力データを上書きする### 8.2 エラーハンドリング- 一部サーバのログ欠損時でも処理を継続可能とする- Glueジョブ失敗時は再実行可能な設計とする---## 9. 補足- 本データは botアクセス可視化を目的とし、人間アクセスは出力対象外とする- 後続処理（BigQuery / Looker Studio）での集計・分析を前提に、 CloudFrontログ形式に近い項目構成としている スクリプト: bot_access_to_s3 ほぼほぼAIの生成結果です。 要件伝えるための壁打ちはある程度必要でしたが、 コーディングに関してはほとんどお任せできました。 import sysimport reimport loggingfrom datetime import datetime, timedelta, timezoneimport boto3from pyspark.context import SparkContextfrom pyspark.sql import functions as Ffrom pyspark.sql.types import ( StructType, StructField, StringType)from awsglue.context import GlueContextfrom awsglue.job import Job# ----------------------------# Optional args parser (GlueのgetResolvedOptionsは必須引数に弱いので自前)# ----------------------------def get_arg(name: str, default=None): """ Glueの引数は --key value の形式で渡される前提 未指定なら default を返す """ key = f"--{name}" if key in sys.argv: idx = sys.argv.index(key) if idx + 1 < len(sys.argv): return sys.argv[idx + 1] return defaultdef parse_s3_uri(uri: str): """ s3://bucket/prefix... を (bucket, prefix) に分解 """ if not uri.startswith("s3://"): raise ValueError(f"Invalid S3 URI: {uri}") no_scheme = uri[5:] parts = no_scheme.split("/", 1) bucket = parts[0] prefix = parts[1] if len(parts) > 1 else "" return bucket, prefixdef ensure_trailing_slash(s: str) -> str: return s if s.endswith("/") else s + "/"# ----------------------------# Normalize helpers# ----------------------------_HYPHENS = [ "\u2010", "\u2011", "\u2012", "\u2013", "\u2014", "\u2212"]def normalize_hyphen(s: str) -> str: if s is None: return None for h in _HYPHENS: s = s.replace(h, "-") return s# ----------------------------# Logging# ----------------------------def setup_logger(level: str): logger = logging.getLogger("bot-access-glue") logger.setLevel(level.upper() if level else "INFO") handler = logging.StreamHandler(sys.stdout) fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s") handler.setFormatter(fmt) # Glueは二重にハンドラが付くことがあるのでガード if not logger.handlers: logger.addHandler(handler) return logger# ----------------------------# S3 prefix exists (boto3)# ----------------------------def s3_prefix_exists(s3_client, bucket: str, prefix: str) -> bool: """ prefix配下に1件でもオブジェクトがあれば True """ resp = s3_client.list_objects_v2(Bucket=bucket, Prefix=prefix, MaxKeys=1) return "Contents" in resp# ----------------------------# Main# ----------------------------def main(): # ---- Parameters ---- # 必須 input_s3_base_uri = get_arg("input_s3_base_uri", None) # 例: s3://xxxxxxxxxxxxxxx/ec2-log/ output_s3_base_uri = get_arg("output_s3_base_uri", None) # 例: s3://xxxxxxxxxxxxxxx/group-a/ target_token = get_arg("target_token", None) # 例: "GPTBot,Claude-User,..." # 任意 target_date = get_arg("target_date", None) # "YYYY-MM-DD" / 未指定なら前日 dry_run = (get_arg("dry_run", "false").lower() == "true") output_mode = get_arg("output_mode", "overwrite") # overwrite | append log_level = get_arg("log_level", "INFO") # INFO | DEBUG max_server_count = int(get_arg("max_server_count", "8")) # デフォ8 output_coalesce = int(get_arg("output_coalesce", "0")) # 0:そのまま / 1:単一ファイル化（小規模向け） bot_type = get_arg("bot_type", "ai") # ai | google_mobile logger = setup_logger(log_level) # validate required if not input_s3_base_uri or not output_s3_base_uri or not target_token: raise ValueError( "Missing required args. " "Required: --input_s3_base_uri --output_s3_base_uri --target_token" ) input_s3_base_uri = ensure_trailing_slash(input_s3_base_uri) output_s3_base_uri = ensure_trailing_slash(output_s3_base_uri) # target_date default: yesterday (JST) if not target_date: # Glue実行リージョンが東京想定なので JST(UTC+9)で「前日」を作る jst = timezone(timedelta(hours=9)) target_date = (datetime.now(jst) - timedelta(days=1)).strftime("%Y-%m-%d") # parse date parts dt = datetime.strptime(target_date, "%Y-%m-%d") yyyy = dt.strftime("%Y") mm = dt.strftime("%m") dd = dt.strftime("%d") # parse tokens # カンマ区切り、前後空白除去、小文字化、ハイフン正規化 raw_tokens = [t.strip() for t in target_token.split(",") if t.strip()] tokens = [normalize_hyphen(t).lower() for t in raw_tokens] logger.info(f"Job started. target_date={target_date} dry_run={dry_run} output_mode={output_mode}") logger.info(f"Bot token count={len(tokens)} max_server_count={max_server_count}") logger.info(f"bot_type={bot_type}") # ---- Discover existing input prefixes ---- s3 = boto3.client("s3") in_bucket, in_base_prefix = parse_s3_uri(input_s3_base_uri) # base_prefix例: "ec2-log/" in_base_prefix = ensure_trailing_slash(in_base_prefix) existing_paths = [] existing_servers = [] missing_servers = [] # 想定命名: prod-webap01 ... prod-webap08 for i in range(1, max_server_count + 1): server = f"prod-webap{i:02d}" # 入力の実際パス: s3://xxxxxxxxxxxxxxx/ec2-log/prod-webap01/apache.access/yyyy/MM/dd/hh/ # Sparkで日単位取得したいので dd/ までを prefix にする（hh配下をまとめて拾う） prefix = f"{in_base_prefix}{server}/apache.access/{yyyy}/{mm}/{dd}/" if log_level.upper() == "DEBUG": logger.debug(f"Checking prefix s3://{in_bucket}/{prefix}") if s3_prefix_exists(s3, in_bucket, prefix): existing_paths.append(f"s3://{in_bucket}/{prefix}") existing_servers.append(server) else: missing_servers.append(server) if not existing_paths: logger.error("No input logs found for any server. job aborted.") raise Exception(f"No input logs found for target_date={target_date}") logger.info(f"Found input paths={len(existing_paths)} servers={existing_servers}") if missing_servers: logger.warning(f"No logs found for servers={missing_servers}") # ---- Glue / Spark init ---- sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) spark = glueContext.spark_session job = Job(glueContext) job.init("bot-access-glue-job", {}) # job nameはGlue側で上書きされてもOK # ---- Read TSV (3 columns) ---- # 形式: collected_at \t source \t json_str df_raw = ( spark.read .option("sep", "\t") .option("quote", "\u0000") # quote無効化（JSON中にquoteがあっても壊れにくくする） .option("escape", "\u0000") # escape無効化 .option("recursiveFileLookup", "true") # 対象パスの配下も再帰的にチェック .csv(existing_paths) .toDF("collected_at", "source", "json_str") ) # ---- Parse JSON (from_json) ---- json_schema = StructType([ StructField("host", StringType(), True), StructField("user", StringType(), True), StructField("time", StringType(), True), StructField("method", StringType(), True), StructField("path", StringType(), True), StructField("code", StringType(), True), StructField("size", StringType(), True), StructField("referer", StringType(), True), StructField("agent", StringType(), True), StructField("proto", StringType(), True), ]) df = df_raw.withColumn("log", F.from_json(F.col("json_str"), json_schema)) # ---- Normalize UA + token matching ---- # UA: 小文字化 + ハイフン正規化 # Spark側で置換する（簡易に代表的なハイフン類を "-" に寄せる） ua = F.col("log.agent") ua_norm = F.lower(ua) for h in _HYPHENS: ua_norm = F.regexp_replace(ua_norm, re.escape(h), "-") df = df.withColumn("ua_norm", ua_norm) cond = F.lit(False) for t in tokens: cond = cond | F.col("ua_norm").contains(t) # Googleボットはtarget_tokenで指定した文字列"Googlebot"に加えて"Mobile"も部分一致するものとする。 if bot_type == "google_mobile": cond = cond & F.col("ua_norm").contains("mobile") df_bot = df.filter(cond) # ---- Timestamp parse (collected_at優先) ---- # 例: 2026-04-19T20:00:06+09:00 ts = F.to_timestamp(F.col("collected_at"), "yyyy-MM-dd'T'HH:mm:ssXXX") df_bot = df_bot.withColumn("ts", ts) # ---- Map required output fields ---- # c-ip: hostの先頭（"ip1, ip2" -> ip1） client_ip = F.when( F.col("log.host").isNull(), F.lit(None) ).otherwise( F.trim(F.split(F.col("log.host"), ",")[0]) ) # referer "-" は NULL referer = F.when((F.col("log.referer") == "-") | (F.col("log.referer") == ""), F.lit(None)) \ .otherwise(F.col("log.referer")) df_out = df_bot.select( F.date_format(F.col("ts"), "yyyy-MM-dd").alias("date"), F.date_format(F.col("ts"), "HH:mm:ss").alias("time"), # F.lit(None).cast(StringType()).alias("x-edge-location"), client_ip.alias("c-ip"), F.col("log.method").alias("cs-method"), # F.lit(None).cast(StringType()).alias("cs(Host)"), F.col("log.path").alias("cs-uri-stem"), F.col("log.code").cast("int").alias("sc-status"), referer.alias("cs(Referer)"), F.col("log.agent").alias("cs(User-Agent)"), # F.lit(None).cast(StringType()).alias("time-taken"), ) # ---- Metrics / counts (高コストなのでDEBUG or dry_run時だけ) ---- if dry_run or log_level.upper() == "DEBUG": read_cnt = df_raw.count() bot_cnt = df_out.count() logger.info(f"Records read={read_cnt}") logger.info(f"Records matched(bot)={bot_cnt}") if bot_cnt == 0: logger.warning(f"No bot traffic found for target_date={target_date}") # ---- Output path ---- out_bucket, out_base_prefix = parse_s3_uri(output_s3_base_uri) out_base_prefix = ensure_trailing_slash(out_base_prefix) # 要件: s3://.../group-a/{yyyy}/{MM}/{dd}/ out_prefix = f"{out_base_prefix}{yyyy}/{mm}/{dd}/" output_path = f"s3://{out_bucket}/{out_prefix}" logger.info(f"Output path={output_path}") # ---- Write ---- if dry_run: logger.info("dry_run=true -> skip writing output.") else: writer = df_out if output_coalesce == 1: writer = writer.coalesce(1) ( writer.write .mode(output_mode) # overwrite | append .option("header", True) # ヘッダあり .csv(output_path) ) logger.info("Job completed successfully") job.commit()if __name__ == "__main__": main() Glueトリガー設定(パラメータ定義) トリガーで以下のパラメータを指定してジョブを実行します。 主にtarget_token、output_s3_base_uriを微修正して、 googleボット、aiボットそれぞれのパラメータをプリセットしておき実行しています。 target_date: 未指定だったらデフォルトで前日、指定あったら特定日付で実行もできるようにtarget_token: 対象botの追加修正にも対応できるように、カンマ区切りで複数入れた文字列input_s3_base_uri: それ以降のサーバーや日付ごとのパスはスクリプト側で実装する想定output_s3_base_uri: target_tokenと合わせて別のトークングループが追加されたときに使う想定dry_run: false | trueoutput_mode: overwrite | appendlog_level: DEBUG | INFObot_type: ai | google_mobile googleボット用トリガー: daily_google_bot_access_to_s3 aiボット用トリガー: daily_ai_bot_access_to_s3 クロスクラウドデータ連携: AWS→Google Google側(連携先) Google側はStorage Transfer Serviceのマニュアル( 参考記事① )や、 データ転送のデベロッパーブログ( 参考記事② )を参考にしていただければと思います。 プロジェクト作成 こちらはスコープ外なので作成済を前提として割愛させていただきます。 Cloud Storage バケット作成 こちらはスコープ外なので作成済を前提として割愛させていただきます。 サブジェクトIDの取得 こちらが今回の連携において肝で、Google側で事前に確認しておく必要があります。 Subject ID取得は以下リンク(googleServiceAccounts.get - Try this method)から可能です。 https://cloud.google.com/storage-transfer/docs/reference/rest/v1/googleServiceAccounts/get 確認した結果をAWS側に連携します。 AWS側(連携元) OIDC 設定(IDプロバイダ) AWSと初めて連携する場合はこれも必要になります。 ここで対象者にサブジェクトIDを入れる必要があるので要注意です。 私はここで少しハマりました... プロバイダ: accounts.google.com対象者: Google側のSubject IDとsts.amazonaws.com The fix is to add the audience you're using to the IdentityProvider. 引用: ( 参考記事③ ) IAMポリシー: GoogleLinkPolicy Roleで参照するポリシーです。 対象のS3に関する権限のみをつけています。 { "Version": "2012-10-17", "Statement": [ { "Sid": "AllowReadOnlyAccessForGoogleStorageTransfer", "Effect": "Allow", "Action": [ "s3:GetObject", "s3:ListBucket" ], "Resource": [ "arn:aws:s3:::xxxxxxxxxxxxxxx", "arn:aws:s3:::xxxxxxxxxxxxxxx/*" ] } ]} IAM Role 作成（信頼ポリシー） 連携・認証用のRoleです。 ここにももちろんサブジェクトIDを入れる必要があります。 上で作成したIDプロバイダのARNも紐づけます。 { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Federated": "arn:aws:iam::xxxxxxxxxxxxxxx:oidc-provider/accounts.google.com" }, "Action": "sts:AssumeRoleWithWebIdentity", "Condition": { "StringEquals": { "accounts.google.com:sub": "xxxxxxxxxxxxxxx" } } } ]} Google側(連携先) ここまでの設定がうまくいっていればあとは簡単、 Storage Transfer Service（STS）にてストレージ転送ジョブの構成を定義するだけです。 デベロッパーブログ( 参考記事② )を参考に設定いただければと思います。 以上となります。 長くなりましたが最後まで読んでいただいた方、ありがとうございます。 別記事リンク 参考記事①: Googleドキュメント Storage Transfer Service https://docs.cloud.google.com/storage-transfer/docs/overview?hl=ja 参考記事②: Googleデベロッパーブログ Storage Transfer Service を使用して AWS から GCP にデータを転送する https://cloud.google.com/blog/ja/topics/developers-practitioners/transfer-data-aws-gcp-using-storage-transfer-service/ 参考記事③: AWSナレッジセンター How do I resolve the AWS STS AssumeRoleWithWebIdentity API call error "InvalidIdentityToken"? https://repost.aws/knowledge-center/iam-sts-invalididentitytoken