265 lines
11 KiB
Python
265 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
SSH 白名单自动更新路由（阿里云安全组）。
|
||
|
||
背景：家庭宽带公网 IP 经常变化，安全组 22 端口若只放行固定 IP，
|
||
换 IP 后就会把自己挡在门外。本路由让家庭电脑定时上报当前公网 IP，
|
||
服务器调用阿里云 ECS API 自动更新安全组 22 端口的入方向白名单。
|
||
|
||
接口（统一响应格式 {success, data, message}）：
|
||
- POST /api/ipwatch/report 上报当前公网 IP，自动增删安全组 22 端口规则
|
||
- GET /api/ipwatch/status 查看服务配置状态（不含任何密钥）
|
||
|
||
安全设计：
|
||
- AccessKey 只存在于服务器 .env，家庭端只持有 IPWATCH_SECRET 上报密钥
|
||
- 上报密钥用 hmac 恒定时间比较，并按来源 IP 限流（默认 60 秒一次）
|
||
- 更新顺序“先加新规则、后删旧规则”，任何一步失败都不会让 SSH 完全断连
|
||
- 只操作“tcp 22/22 且来源为单个 IP(/32)”的规则，绝不碰 0.0.0.0/0 等其它规则
|
||
"""
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import ipaddress
|
||
import json
|
||
import logging
|
||
import os
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
||
from pydantic import BaseModel, Field
|
||
|
||
from ..schemas import UnifiedResponse
|
||
from ..security import RateLimiter, get_client_ip
|
||
|
||
router = APIRouter(prefix="/api/ipwatch", tags=["ipwatch"])
|
||
|
||
# 日志走 uvicorn 的 logger，方便 journalctl 查看
|
||
logger = logging.getLogger("uvicorn.error")
|
||
|
||
|
||
# ---------- 环境变量配置（服务器 .env，见 .env.example 中文注释） ----------
|
||
|
||
IPWATCH_SECRET = (os.getenv("IPWATCH_SECRET") or "").strip()
|
||
ALIYUN_AK_ID = (os.getenv("ALIYUN_AK_ID") or "").strip()
|
||
ALIYUN_AK_SECRET = (os.getenv("ALIYUN_AK_SECRET") or "").strip()
|
||
SECURITY_GROUP_ID = (os.getenv("IPWATCH_SECURITY_GROUP_ID") or "").strip()
|
||
IPWATCH_REGION = (os.getenv("IPWATCH_REGION") or "cn-hangzhou").strip()
|
||
IPWATCH_PORT = int(os.getenv("IPWATCH_PORT") or "22")
|
||
|
||
# 同一来源 IP 两次上报的最小间隔（秒），防止密钥泄露后被刷白名单
|
||
try:
|
||
_report_interval = int(os.getenv("IPWATCH_REPORT_INTERVAL_SECONDS") or "60")
|
||
except ValueError:
|
||
_report_interval = 60
|
||
report_limiter = RateLimiter(1, max(_report_interval, 1))
|
||
|
||
|
||
class IpWatchReport(BaseModel):
|
||
"""上报请求体：家庭端上传密钥与当前公网 IP。"""
|
||
|
||
secret: str = Field(min_length=1, max_length=256)
|
||
ip: str = Field(min_length=7, max_length=45)
|
||
|
||
|
||
def _aliyun_call(action: str, params: dict) -> dict:
|
||
"""调用阿里云 ECS RPC API（HMAC-SHA1 签名，仅用 Python 标准库）。
|
||
|
||
Aliyun OpenAPI 签名规则：对全部参数按 key 排序后拼接，
|
||
再用 AccessKeySecret 做 HMAC-SHA1，最后 BASE64 得到 Signature。
|
||
"""
|
||
query = {
|
||
"AccessKeyId": ALIYUN_AK_ID,
|
||
"Action": action,
|
||
"Format": "JSON",
|
||
"SignatureMethod": "HMAC-SHA1",
|
||
"SignatureNonce": uuid.uuid4().hex, # 每次请求唯一，防止重放
|
||
"SignatureVersion": "1.0",
|
||
"Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||
"Version": "2014-05-26",
|
||
"RegionId": IPWATCH_REGION,
|
||
}
|
||
query.update(params or {})
|
||
|
||
def _enc(value: str) -> str:
|
||
"""RFC3986 百分号编码（阿里云要求保留 -_.~ 三个字符）。"""
|
||
return urllib.parse.quote(str(value), safe="-_.~")
|
||
|
||
canonical = "&".join(f"{_enc(k)}={_enc(v)}" for k, v in sorted(query.items()))
|
||
string_to_sign = "GET&%2F&" + _enc(canonical)
|
||
signature = base64.b64encode(
|
||
hmac.new((ALIYUN_AK_SECRET + "&").encode(), string_to_sign.encode(), hashlib.sha1).digest()
|
||
).decode()
|
||
url = f"https://ecs.{IPWATCH_REGION}.aliyuncs.com/?{canonical}&Signature={_enc(signature)}"
|
||
|
||
request = urllib.request.Request(url, headers={"User-Agent": "MyBlog-ipwatch/1.0"})
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=20) as resp:
|
||
return json.loads(resp.read().decode())
|
||
except urllib.error.HTTPError as exc:
|
||
# 阿里云返回错误时，响应体里是 JSON 格式的错误信息
|
||
try:
|
||
return json.loads(exc.read().decode())
|
||
except Exception:
|
||
return {"Code": "HTTP_ERROR", "Message": f"阿里云接口返回 HTTP {exc.code}"}
|
||
except Exception as exc:
|
||
return {"Code": "NETWORK_ERROR", "Message": f"无法连接阿里云接口：{exc}"}
|
||
|
||
|
||
def _fetch_ingress_rules() -> list:
|
||
"""读取安全组全部入方向规则，失败时抛出 502。"""
|
||
resp = _aliyun_call(
|
||
"DescribeSecurityGroupAttribute",
|
||
{"SecurityGroupId": SECURITY_GROUP_ID, "Direction": "ingress"},
|
||
)
|
||
if resp.get("Code"):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||
detail=f"读取安全组规则失败：{resp.get('Message') or resp.get('Code')}",
|
||
)
|
||
return resp.get("Permissions", {}).get("Permission", [])
|
||
|
||
|
||
def _host_ip(cidr: str) -> str:
|
||
"""把安全组来源归一化为纯 IP；非 IPv4 单 IP(/32) 返回空字符串。
|
||
|
||
阿里云返回的 SourceCidrIp 可能是 "1.2.3.4" 或 "1.2.3.4/32"，
|
||
本函数只认 IPv4 的 /32 单 IP，其它（0.0.0.0/0、IPv6 等）一律忽略，
|
||
防止误删用户手动配置的放行规则。
|
||
"""
|
||
if not cidr:
|
||
return ""
|
||
value = cidr.strip()
|
||
if "/" not in value:
|
||
try:
|
||
ip = ipaddress.ip_address(value)
|
||
except ValueError:
|
||
return ""
|
||
return str(ip) if ip.version == 4 else ""
|
||
try:
|
||
network = ipaddress.ip_network(value, strict=False)
|
||
except ValueError:
|
||
return ""
|
||
if network.version != 4 or network.prefixlen != 32:
|
||
return ""
|
||
return str(network.network_address)
|
||
|
||
|
||
@router.get("/status", response_model=UnifiedResponse)
|
||
def status_info() -> UnifiedResponse:
|
||
"""查看服务配置状态（不返回任何密钥），便于部署后排查。"""
|
||
data = {
|
||
"configured": bool(IPWATCH_SECRET and ALIYUN_AK_ID and ALIYUN_AK_SECRET and SECURITY_GROUP_ID),
|
||
"region": IPWATCH_REGION,
|
||
"security_group_id": SECURITY_GROUP_ID,
|
||
"port": IPWATCH_PORT,
|
||
"report_interval_seconds": report_limiter.window_seconds,
|
||
}
|
||
return UnifiedResponse(success=True, data=data, message="获取成功")
|
||
|
||
|
||
@router.post("/report", response_model=UnifiedResponse)
|
||
def report(payload: IpWatchReport, request: Request) -> UnifiedResponse:
|
||
"""上报当前公网 IP：先加新白名单、再删旧白名单，全程不会锁死 SSH。"""
|
||
# 1. 服务配置检查（未配置时直接拒绝，避免密钥为空被绕过）
|
||
if not IPWATCH_SECRET:
|
||
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="IP 白名单服务未配置")
|
||
if not (ALIYUN_AK_ID and ALIYUN_AK_SECRET):
|
||
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="阿里云密钥未配置")
|
||
if not SECURITY_GROUP_ID:
|
||
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="安全组未配置")
|
||
|
||
# 2. 上报密钥校验（恒定时间比较，防止时序侧信道）
|
||
if not hmac.compare_digest(payload.secret.encode("utf-8"), IPWATCH_SECRET.encode("utf-8")):
|
||
logger.warning("ipwatch: 上报密钥错误，来源 %s", get_client_ip(request))
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="上报密钥错误")
|
||
|
||
# 3. 按来源 IP 限流，防止密钥泄露后被无限刷白名单
|
||
client_ip = get_client_ip(request)
|
||
if report_limiter.is_blocked(client_ip):
|
||
raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="上报过于频繁，请稍后再试")
|
||
|
||
# 4. IP 格式校验（仅支持 IPv4）
|
||
ip_text = payload.ip.strip()
|
||
try:
|
||
ip_obj = ipaddress.ip_address(ip_text)
|
||
except ValueError:
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="IP 地址格式不正确")
|
||
if ip_obj.version != 4:
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="仅支持 IPv4 地址")
|
||
new_ip = str(ip_obj)
|
||
new_cidr = f"{new_ip}/32"
|
||
target_range = f"{IPWATCH_PORT}/{IPWATCH_PORT}"
|
||
|
||
# 5. 读取当前 22 端口规则，判断是否需要更新
|
||
rules = _fetch_ingress_rules()
|
||
port_rules = [
|
||
r
|
||
for r in rules
|
||
if str(r.get("IpProtocol", "")).lower() == "tcp" and r.get("PortRange") == target_range
|
||
]
|
||
already_authorized = any(_host_ip(r.get("SourceCidrIp")) == new_ip for r in port_rules)
|
||
if already_authorized:
|
||
return UnifiedResponse(
|
||
success=True,
|
||
data={"changed": False, "ip": new_ip, "port": IPWATCH_PORT},
|
||
message="IP 已在白名单中，无需更新",
|
||
)
|
||
|
||
# 6. 先添加新 IP 规则（成功后才进入删除阶段，避免 SSH 断连）
|
||
add_resp = _aliyun_call(
|
||
"AuthorizeSecurityGroup",
|
||
{
|
||
"SecurityGroupId": SECURITY_GROUP_ID,
|
||
"IpProtocol": "tcp",
|
||
"PortRange": target_range,
|
||
"SourceCidrIp": new_cidr,
|
||
"Policy": "accept",
|
||
"Description": "MyBlog IPWatch 自动白名单",
|
||
},
|
||
)
|
||
if add_resp.get("Code"):
|
||
logger.error("ipwatch: 添加白名单失败 %s -> %s", new_cidr, add_resp)
|
||
raise HTTPException(
|
||
status_code=status.HTTP_502_BAD_GATEWAY,
|
||
detail=f"添加白名单失败：{add_resp.get('Message') or add_resp.get('Code')}",
|
||
)
|
||
|
||
# 7. 删除旧的单 IP 规则（只删 22 端口 /32 来源，绝不动 0.0.0.0/0 等规则）
|
||
removed = []
|
||
for rule in port_rules:
|
||
old_ip = _host_ip(rule.get("SourceCidrIp"))
|
||
if not old_ip or old_ip == new_ip:
|
||
continue
|
||
# 阿里云对旧规则的存储格式不统一（有的带 /32，有的不带），
|
||
# 依次尝试两种格式；任一成功即视为删除完成
|
||
revoke_ok = False
|
||
for cidr_candidate in (f"{old_ip}/32", old_ip):
|
||
revoke_resp = _aliyun_call(
|
||
"RevokeSecurityGroup",
|
||
{
|
||
"SecurityGroupId": SECURITY_GROUP_ID,
|
||
"IpProtocol": "tcp",
|
||
"PortRange": target_range,
|
||
"SourceCidrIp": cidr_candidate,
|
||
"Policy": "accept",
|
||
},
|
||
)
|
||
if not revoke_resp.get("Code"):
|
||
revoke_ok = True
|
||
break
|
||
if revoke_ok:
|
||
removed.append(old_ip)
|
||
else:
|
||
# 新规则已生效，旧规则删除失败只是残留，不影响 SSH 可用性
|
||
logger.warning("ipwatch: 删除旧规则失败 %s -> %s", old_ip, revoke_resp)
|
||
|
||
report_limiter.hit(client_ip)
|
||
data = {"changed": True, "ip": new_ip, "port": IPWATCH_PORT, "removed": removed}
|
||
logger.info("ipwatch: 白名单更新完成 新IP=%s 删除=%s", new_ip, removed)
|
||
return UnifiedResponse(success=True, data=data, message="白名单更新成功")
|