#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ec2user_mail_sender.py
EC2個人OSユーザー作成後のメール作成補助。
MODE:
TEST
- 登録者情報.xlsx を読み込む
- 「環境 + ユーザー名」でまとめる
- PEM存在確認
- PEMから公開鍵を取得できることを確認
- PEMを使って「新規ユーザーID@対象IP」にSSH接続し、whoamiを確認
- 作成予定メール（To/CC/件名/本文/添付）を表示
- Outlookは操作しない
EXECUTE
- TESTと同じ確認を再実施
- 全確認OKのグループだけOutlookメールを作成
- PEMを添付
- メール画面を表示
- 絶対に自動送信しない（Sendは呼ばない）
前提:
OS_user/
ec2user_mail_sender.py
ec2user/
input/
登録者情報.xlsx
output/
log/
pemkey/
<ユーザー名>/
<環境>/
<環境>-9055-tyo-keypair-<ユーザー名>-<IP末尾>.pem
"""
from __future__ import annotations
import re
import shutil
import subprocess
import sys
import time
import unicodedata
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Optional
# ============================================================
# 設定
# ============================================================
MODE = "TEST" # "TEST" / "EXECUTE"
BASE_DIR = Path(__file__).resolve().parent
EC2USER_DIR = BASE_DIR / "ec2user"
INPUT_DIR = EC2USER_DIR / "input"
OUTPUT_DIR = EC2USER_DIR / "output"
PEM_ROOT = OUTPUT_DIR / "pemkey"
LOG_DIR = OUTPUT_DIR / "log"
REGISTER_EXCEL = INPUT_DIR / "登録者情報.xlsx"
# 添付画像の手順に合わせたCC
MAIL_CC = "asahi_data_connect@n-ais.co.jp"
# Outlook表示間隔。複数メールを作る場合に少し待つ。
OUTLOOK_DISPLAY_INTERVAL_SEC = 0.5
# SSH
SSH_CONNECT_TIMEOUT = 10
# Excel列（1始まり）
COL_REGISTER = 1 # A: 登録用
COL_ENVIRONMENT = 2 # B: 環境
COL_IP = 3 # C: IPアドレス
COL_USER_ID = 4 # D: ユーザーID
COL_USER_NAME = 5 # E: ユーザー名
COL_EMAIL = 7 # G: メールアドレス（ユーザー指定）
# ============================================================
# データ
# ============================================================
@dataclass(frozen=True)
class Target:
excel_row: int
environment: str
ip: str
user_id: str
user_name: str
email: str
pem_path: Path
@dataclass
class CheckResult:
target: Target
ok: bool
messages: list[str]
_LOG_FILE: Optional[Path] = None
# ============================================================
# 共通
# ============================================================
def log_print(message: str = "") -> None:
print(message)
if _LOG_FILE is not None:
try:
with _LOG_FILE.open("a", encoding="utf-8") as f:
f.write(message + "\n")
except OSError:
pass
def ok(message: str) -> str:
return f"[OK] {message}"
def ng(message: str) -> str:
return f"[NG] {message}"
def normalize(value) -> str:
if value is None:
return ""
return unicodedata.normalize("NFKC", str(value)).strip()
def is_register_target(value) -> bool:
return normalize(value).upper() in {"Y", "YES", "1", "TRUE", "○", "〇"}
def validate_ip(ip: str) -> bool:
parts = ip.split(".")
if len(parts) != 4:
return False
try:
nums = [int(x) for x in parts]
except ValueError:
return False
return all(0 <= x <= 255 for x in nums)
def ip_last_octet(ip: str) -> str:
return ip.rsplit(".", 1)[-1]
def expected_pem_path(environment: str, user_name: str, ip: str) -> Path:
filename = (
f"{environment}-9055-tyo-keypair-"
f"{user_name}-{ip_last_octet(ip)}.pem"
)
return PEM_ROOT / user_name / environment / filename
def run_process(args: list[str], timeout: int = 20) -> subprocess.CompletedProcess:
return subprocess.run(
args,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=timeout,
shell=False,
)
# ============================================================
# Excel
# ============================================================
def load_targets() -> list[Target]:
"""
openpyxlがあればそれを利用。
会社PCで追加インストール不可の場合に備え、未導入なら明確にエラー表示。
"""
try:
import openpyxl
except ImportError:
raise RuntimeError(
"openpyxl が見つかりません。"
"現在の ec2user_generator.py と同じExcel読込方式を使用する場合は、"
"openpyxl が利用可能な環境で実行してください。"
)
if not REGISTER_EXCEL.exists():
raise FileNotFoundError(f"登録者情報.xlsx が見つかりません: {REGISTER_EXCEL}")
wb = openpyxl.load_workbook(REGISTER_EXCEL, data_only=True, read_only=True)
try:
ws = wb.active
targets: list[Target] = []
# 1行目は見出し想定
for row in range(2, ws.max_row + 1):
register = ws.cell(row, COL_REGISTER).value
if not is_register_target(register):
continue
env = normalize(ws.cell(row, COL_ENVIRONMENT).value).lower()
ip = normalize(ws.cell(row, COL_IP).value)
user_id = normalize(ws.cell(row, COL_USER_ID).value)
user_name = normalize(ws.cell(row, COL_USER_NAME).value)
email = normalize(ws.cell(row, COL_EMAIL).value)
# 完全空行は無視
if not any([env, ip, user_id, user_name, email]):
continue
pem = expected_pem_path(env, user_name, ip) if env and user_name and ip else Path()
targets.append(
Target(
excel_row=row,
environment=env,
ip=ip,
user_id=user_id,
user_name=user_name,
email=email,
pem_path=pem,
)
)
return targets
finally:
wb.close()
# ============================================================
# SSH / PEMチェック
# ============================================================
def derive_public_key(pem_path: Path) -> tuple[bool, str]:
try:
proc = run_process(
["ssh-keygen", "-y", "-f", str(pem_path)],
timeout=15,
)
except Exception as exc:
return False, str(exc)
if proc.returncode != 0:
return False, (proc.stderr.strip() or proc.stdout.strip())
public_key = proc.stdout.strip()
if not public_key.startswith(("ssh-ed25519 ", "ssh-rsa ", "ecdsa-")):
return False, "公開鍵として認識できない出力です。"
return True, public_key
def ssh_whoami(target: Target) -> tuple[bool, str]:
"""
新規ユーザー用PEMを使って、実際に新規ユーザーでSSHログインする。
サーバー側の変更は一切行わない。
"""
args = [
"ssh",
"-i", str(target.pem_path),
"-o", "BatchMode=yes",
"-o", f"ConnectTimeout={SSH_CONNECT_TIMEOUT}",
"-o", "StrictHostKeyChecking=accept-new",
"-o", "PasswordAuthentication=no",
f"{target.user_id}@{target.ip}",
"whoami",
]
try:
proc = run_process(args, timeout=SSH_CONNECT_TIMEOUT + 10)
except subprocess.TimeoutExpired:
return False, "SSH接続がタイムアウトしました。"
except Exception as exc:
return False, str(exc)
if proc.returncode != 0:
detail = proc.stderr.strip() or proc.stdout.strip()
return False, detail
actual = proc.stdout.strip().splitlines()
actual_user = actual[-1].strip() if actual else ""
if actual_user != target.user_id:
return False, f"whoami={actual_user!r}（期待値={target.user_id!r}）"
return True, actual_user
def check_target(target: Target) -> CheckResult:
messages: list[str] = []
good = True
if not target.environment:
messages.append(ng("環境が空欄です。"))
good = False
if not target.ip or not validate_ip(target.ip):
messages.append(ng(f"IPアドレスが不正です: {target.ip!r}"))
good = False
if not target.user_id:
messages.append(ng("ユーザーIDが空欄です。"))
good = False
if not target.user_name:
messages.append(ng("ユーザー名が空欄です。"))
good = False
if not target.email:
messages.append(ng("メールアドレス（G列）が空欄です。"))
good = False
elif "@" not in target.email:
messages.append(ng(f"メールアドレス形式を確認してください: {target.email}"))
good = False
if not good:
return CheckResult(target, False, messages)
if not target.pem_path.is_file():
messages.append(ng(f"PEMが見つかりません: {target.pem_path}"))
return CheckResult(target, False, messages)
messages.append(ok(f"PEM存在: {target.pem_path.name}"))
pub_ok, pub_detail = derive_public_key(target.pem_path)
if not pub_ok:
messages.append(ng(f"PEMから公開鍵を取得できません: {pub_detail}"))
return CheckResult(target, False, messages)
messages.append(ok("PEMから公開鍵取得可能"))
ssh_ok, ssh_detail = ssh_whoami(target)
if not ssh_ok:
messages.append(
ng(
f"PEMを使用したSSHログイン確認NG: "
f"{target.user_id}@{target.ip} / {ssh_detail}"
)
)
return CheckResult(target, False, messages)
messages.append(
ok(
f"PEMを使用したSSHログイン成功: "
f"{target.user_id}@{target.ip} / whoami={ssh_detail}"
)
)
return CheckResult(target, True, messages)
# ============================================================
# グループ化・メール
# ============================================================
def group_targets(targets: list[Target]) -> dict[tuple[str, str], list[Target]]:
"""
同じ「環境 + ユーザー名」を1通にまとめる。
testとdevは必ず別メール。
"""
groups: dict[tuple[str, str], list[Target]] = {}
for target in targets:
key = (target.environment, target.user_name)
groups.setdefault(key, []).append(target)
# IPで安定ソート
for values in groups.values():
values.sort(key=lambda t: tuple(int(x) for x in t.ip.split(".")) if validate_ip(t.ip) else (999,))
return groups
def validate_group_consistency(targets: list[Target]) -> tuple[bool, list[str]]:
"""
同一メールにまとめる対象で、ユーザーIDとメールアドレスが一致するか確認。
不一致なら誤送信防止のためメールを作らない。
"""
errors: list[str] = []
user_ids = {t.user_id for t in targets}
emails = {t.email.lower() for t in targets}
if len(user_ids) != 1:
errors.append(
f"同じ環境+ユーザー名なのにユーザーIDが複数あります: {sorted(user_ids)}"
)
if len(emails) != 1:
errors.append(
f"同じ環境+ユーザー名なのにメールアドレスが複数あります: {sorted(emails)}"
)
# 同じPEMを重複添付しないため確認
pem_paths = [str(t.pem_path.resolve()) for t in targets]
if len(pem_paths) != len(set(pem_paths)):
errors.append("同一PEMが重複しています。")
return len(errors) == 0, errors
def build_subject(environment: str) -> str:
return f"【ADC】（{environment}環境）：EC2ユーザー情報の連絡"
def build_body(targets: list[Target]) -> str:
first = targets[0]
ip_lines = "\r\n".join(t.ip for t in targets)
return (
"お疲れ様です。\r\n"
"\r\n"
"EC2個人ユーザーの作成が完了しましたので\r\n"
"IDと秘密鍵を本メールにて送付いたします。\r\n"
"※ログインする際には秘密鍵のみとなるため、"
"パスワードは発行しておりません。\r\n"
"\r\n"
f"ID: {first.user_id}\r\n"
"対象サーバのipアドレス：\r\n"
f"{ip_lines}\r\n"
"\r\n"
"秘密鍵は本メールに添付しています。\r\n"
"\r\n"
"ログイン確認を実施いただき、結果を返信いただけますでしょうか。\r\n"
"よろしくお願いいたします。\r\n"
)
def print_mail_preview(targets: list[Target]) -> None:
first = targets[0]
subject = build_subject(first.environment)
body = build_body(targets)
log_print("-" * 70)
log_print("【作成予定メール】")
log_print(f"グループ : {first.environment} + {first.user_name}")
log_print(f"To : {first.email}")
log_print(f"CC : {MAIL_CC}")
log_print(f"Subject : {subject}")
log_print("")
log_print("本文:")
for line in body.replace("\r", "").split("\n"):
log_print(f" {line}")
log_print("添付:")
for target in targets:
log_print(f" {target.pem_path}")
log_print("-" * 70)
# ============================================================
# Outlook
# ============================================================
def create_outlook_draft(targets: list[Target]) -> tuple[bool, str]:
"""
Outlookデスクトップ版でメールを作成して表示する。
Send() は絶対に呼ばない。
"""
try:
import win32com.client # type: ignore
except ImportError:
return (
False,
"win32com.client が見つかりません。"
"Outlook操作には pywin32 が必要です。"
)
first = targets[0]
try:
outlook = win32com.client.Dispatch("Outlook.Application")
mail = outlook.CreateItem(0) # olMailItem
mail.To = first.email
mail.CC = MAIL_CC
mail.Subject = build_subject(first.environment)
mail.Body = build_body(targets)
for target in targets:
mail.Attachments.Add(str(target.pem_path.resolve()))
# 重要:
# Display() のみ。Send() は実装しない。
mail.Display(False)
return True, "Outlookメールを作成して表示しました（未送信）。"
except Exception as exc:
return False, f"Outlookメール作成に失敗しました: {exc}"
# ============================================================
# main
# ============================================================
def main() -> int:
global _LOG_FILE
mode = MODE.strip().upper()
if mode not in {"TEST", "EXECUTE"}:
print("MODE は TEST または EXECUTE を指定してください。")
return 2
if shutil.which("ssh") is None:
print("[ERROR] ssh.exe が見つかりません。OpenSSH Clientを確認してください。")
return 2
if shutil.which("ssh-keygen") is None:
print("[ERROR] ssh-keygen.exe が見つかりません。OpenSSH Clientを確認してください。")
return 2
LOG_DIR.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
_LOG_FILE = LOG_DIR / f"ec2user_mail_{mode.lower()}_{timestamp}.log"
log_print("=" * 70)
log_print("EC2 OSユーザー メール作成補助")
log_print(f"MODE : {mode}")
log_print(f"開始 : {datetime.now():%Y-%m-%d %H:%M:%S}")
log_print(f"Excel: {REGISTER_EXCEL}")
log_print(f"LOG : {_LOG_FILE}")
log_print("=" * 70)
if mode == "TEST":
log_print(
"TESTモード：PEM/SSH/メール内容を確認します。"
"Outlookメールは作成しません。"
)
else:
log_print(
"EXECUTEモード：全チェックOK後、Outlookメールを表示します。"
)
log_print(
"※ 自動送信はしません。内容を確認して手動で送信してください。"
)
try:
targets = load_targets()
except Exception as exc:
log_print(ng(str(exc)))
return 2
if not targets:
log_print(ng("登録用=Y の対象がありません。"))
return 1
groups = group_targets(targets)
log_print(f"対象行数 : {len(targets)}")
log_print(f"メール単位数 : {len(groups)}")
log_print("")
total_groups = 0
ok_groups = 0
ng_groups = 0
drafts_created = 0
for (environment, user_name), group in groups.items():
total_groups += 1
log_print("=" * 70)
log_print(f"環境 : {environment}")
log_print(f"ユーザー名 : {user_name}")
log_print(f"対象台数 : {len(group)}")
log_print("=" * 70)
consistent, consistency_errors = validate_group_consistency(group)
if not consistent:
for error in consistency_errors:
log_print(ng(error))
log_print(ng("誤送信防止のため、このグループのメールは作成しません。"))
ng_groups += 1
log_print("")
continue
all_ok = True
for target in group:
log_print(
f"[CHECK] Excel行={target.excel_row} "
f"IP={target.ip} ID={target.user_id}"
)
result = check_target(target)
for message in result.messages:
log_print(f" {message}")
if not result.ok:
all_ok = False
if not all_ok:
log_print(ng("1台以上の確認がNGのため、このグループのメールは作成しません。"))
ng_groups += 1
log_print("")
continue
ok_groups += 1
log_print(ok("グループ内の全PEM・SSHログイン確認OK"))
# TESTでも必ず作成予定内容を表示
print_mail_preview(group)
if mode == "EXECUTE":
created, detail = create_outlook_draft(group)
if not created:
log_print(ng(detail))
ng_groups += 1
ok_groups -= 1
log_print("")
continue
drafts_created += 1
log_print(ok(detail))
log_print(
"【重要】メールはまだ送信されていません。"
"宛先・本文・添付PEMを確認してから手動で送信してください。"
)
time.sleep(OUTLOOK_DISPLAY_INTERVAL_SEC)
log_print("")
log_print("=" * 70)
log_print("処理結果")
log_print(f"メール単位 : {total_groups}")
log_print(f"確認OK : {ok_groups}")
log_print(f"確認NG : {ng_groups}")
if mode == "EXECUTE":
log_print(f"Outlook作成: {drafts_created}")
log_print("自動送信 : 0（自動送信機能はありません）")
log_print(f"終了 : {datetime.now():%Y-%m-%d %H:%M:%S}")
log_print("=" * 70)
return 0 if ng_groups == 0 else 1
if __name__ == "__main__":
sys.exit(main())
