# -*- 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="白名单更新成功")