1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
#!/usr/bin/env python3
"""
Route 53 DNS 設定スクリプト: SPF / DKIM / DMARC
使い方（サーバー上で実行）:
pip3 install boto3
export AWS_ACCESS_KEY_ID=...
export AWS_SECRET_ACCESS_KEY=...
python3 /app/infra/mail/dns/setup-route53.py
HOSTED_ZONE_ID は yyamashita.com のゾーンを自動検出する。
SERVER_IP は自動取得（必要に応じて環境変数で上書き可能）。
"""
import os
import re
import sys
import urllib.request
try:
import boto3
except ImportError:
print("pip3 install boto3 を実行してください", file=sys.stderr)
sys.exit(1)
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
DOMAINS_FILE = os.environ.get("DOMAINS_FILE", os.path.join(SCRIPT_DIR, "..", "domains.txt"))
SELECTOR = "mail"
PRIMARY = "hetzner.yyamashita.com"
ZONE_DOMAIN = "yyamashita.com"
def get_server_ip():
if ip := os.environ.get("SERVER_IP", ""):
return ip
for url in ["https://api.ipify.org", "https://ifconfig.me/ip"]:
try:
with urllib.request.urlopen(url, timeout=5) as r:
return r.read().decode().strip()
except Exception:
continue
return ""
def get_hosted_zone_id(client):
if zone_id := os.environ.get("HOSTED_ZONE_ID", ""):
return zone_id
paginator = client.get_paginator("list_hosted_zones")
for page in paginator.paginate():
for zone in page["HostedZones"]:
if zone["Name"].rstrip(".") == ZONE_DOMAIN:
return zone["Id"].split("/")[-1]
return ""
def read_domains(path):
domains = []
with open(path) as f:
for line in f:
line = line.strip()
if line and not line.startswith("#"):
domains.append(line)
return domains
def read_dkim_pubkey(path):
with open(path) as f:
content = f.read()
parts = re.findall(r'"([^"]*)"', content)
return "".join(parts)
def txt_record(fqdn, raw_value):
chunks = [raw_value[i:i + 255] for i in range(0, len(raw_value), 255)]
txt_value = " ".join(f'"{c}"' for c in chunks)
name = fqdn if fqdn.endswith(".") else fqdn + "."
return {
"Action": "UPSERT",
"ResourceRecordSet": {
"Name": name,
"Type": "TXT",
"TTL": 300,
"ResourceRecords": [{"Value": txt_value}],
},
}
def main():
client = boto3.client("route53")
zone_id = get_hosted_zone_id(client)
if not zone_id:
print(f"ERROR: {ZONE_DOMAIN} のホストゾーンが見つかりません", file=sys.stderr)
print(" export HOSTED_ZONE_ID=Z... で手動指定してください", file=sys.stderr)
sys.exit(1)
server_ip = get_server_ip()
if not server_ip:
print("ERROR: SERVER_IP を自動取得できませんでした", file=sys.stderr)
print(" export SERVER_IP=<HetznerサーバーIP>", file=sys.stderr)
sys.exit(1)
dkim_key_file = "/etc/opendkim/keys/mail.txt"
try:
dkim_value = read_dkim_pubkey(dkim_key_file)
except FileNotFoundError:
print(f"ERROR: DKIM 公開鍵ファイルが見つかりません: {dkim_key_file}", file=sys.stderr)
print("先に mail/install.sh を実行してください", file=sys.stderr)
sys.exit(1)
service_domains = read_domains(DOMAINS_FILE)
print(f"サーバー IP : {server_ip}")
print(f"ホストゾーン: {zone_id} ({ZONE_DOMAIN})")
print(f"ドメイン : {[PRIMARY] + service_domains}")
print()
changes = []
# プライマリドメイン (IP 直指定 SPF)
changes.append(txt_record(PRIMARY, f"v=spf1 ip4:{server_ip} -all"))
changes.append(txt_record(f"{SELECTOR}._domainkey.{PRIMARY}", dkim_value))
changes.append(txt_record(f"_dmarc.{PRIMARY}", "v=DMARC1; p=none"))
# サービスドメイン (include SPF)
for domain in service_domains:
changes.append(txt_record(domain, f"v=spf1 include:{PRIMARY} -all"))
changes.append(txt_record(f"{SELECTOR}._domainkey.{domain}", dkim_value))
changes.append(txt_record(f"_dmarc.{domain}", "v=DMARC1; p=none"))
print(f"{len(changes)} 件の DNS レコードを UPSERT します...")
for c in changes:
rrs = c["ResourceRecordSet"]
print(f" {rrs['Type']} {rrs['Name']}")
resp = client.change_resource_record_sets(
HostedZoneId=zone_id,
ChangeBatch={
"Comment": "Mail auth records (SPF/DKIM/DMARC) via hetzner-infra",
"Changes": changes,
},
)
info = resp["ChangeInfo"]
print()
print(f"Change ID : {info['Id']}")
print(f"Status : {info['Status']}")
print()
print("確認コマンド (数分後):")
for domain in [PRIMARY] + service_domains:
print(f" dig TXT {SELECTOR}._domainkey.{domain} +short")
if __name__ == "__main__":
main()
|
