| import os |
| import socket |
| import smtplib |
| import ssl |
| import traceback |
| from email.mime.multipart import MIMEMultipart |
| from email.mime.text import MIMEText |
| from email.header import Header |
|
|
| import gradio as gr |
|
|
|
|
| |
| DEFAULT_SMTP_SERVER = os.getenv("SMTP_SERVER", "smtpauth.intel.com") |
| DEFAULT_SMTP_PORT = int(os.getenv("SMTP_PORT", "587")) |
|
|
| DEFAULT_MAIL_SENDER = os.getenv("MAIL_SENDER", "wenjiao.yue@intel.com") |
| DEFAULT_MAIL_TO = os.getenv("MAIL_TO", "wenjiao.yue@intel.com") |
|
|
| MAIL_PASSWORD = os.getenv("MAIL_PASSWORD") |
|
|
|
|
| def test_dns(smtp_server: str, smtp_port: int): |
| """ |
| 测试 DNS 解析。 |
| 如果这里失败，说明还没到端口连接阶段。 |
| """ |
| try: |
| results = socket.getaddrinfo(smtp_server, smtp_port) |
| formatted = [] |
|
|
| for item in results: |
| family, socktype, proto, canonname, sockaddr = item |
| formatted.append( |
| f"family={family}, socktype={socktype}, proto={proto}, " |
| f"canonname={canonname}, sockaddr={sockaddr}" |
| ) |
|
|
| return True, "DNS 解析成功：\n" + "\n".join(formatted) |
|
|
| except Exception as e: |
| return False, ( |
| f"DNS 解析失败：{repr(e)}\n\n" |
| f"SMTP_SERVER={smtp_server}\n" |
| f"SMTP_PORT={smtp_port}\n\n" |
| "说明 Hugging Face Space 当前无法把这个域名解析成 IP。\n" |
| "这通常不是端口问题，而是域名在公网 DNS 中不可解析，" |
| "或者它依赖公司内网 DNS。" |
| ) |
|
|
|
|
| def test_tcp_connection(smtp_server: str, smtp_port: int): |
| """ |
| 测试 TCP 连接。 |
| DNS 成功后，才会测试到这一步。 |
| """ |
| try: |
| with socket.create_connection((smtp_server, smtp_port), timeout=15) as sock: |
| peer = sock.getpeername() |
|
|
| return True, f"TCP 连接成功：{smtp_server}:{smtp_port}, peer={peer}" |
|
|
| except Exception as e: |
| return False, ( |
| f"TCP 连接失败：{repr(e)}\n\n" |
| "如果 DNS 成功但这里失败，才可能是端口、防火墙、网络策略问题。" |
| ) |
|
|
|
|
| def test_smtp_handshake(smtp_server: str, smtp_port: int): |
| """ |
| 测试 SMTP EHLO 和 STARTTLS。 |
| """ |
| logs = [] |
|
|
| try: |
| with smtplib.SMTP(smtp_server, smtp_port, timeout=30) as smtp: |
| smtp.set_debuglevel(0) |
|
|
| code, msg = smtp.ehlo() |
| logs.append(f"EHLO 返回：code={code}, msg={msg!r}") |
|
|
| if code != 250: |
| return False, "\n".join(logs) + "\n\nEHLO 失败。" |
|
|
| context = ssl._create_unverified_context() |
| code, msg = smtp.starttls(context=context) |
| logs.append(f"STARTTLS 返回：code={code}, msg={msg!r}") |
|
|
| code, msg = smtp.ehlo() |
| logs.append(f"TLS 后 EHLO 返回：code={code}, msg={msg!r}") |
|
|
| return True, "SMTP 握手成功：\n" + "\n".join(logs) |
|
|
| except Exception as e: |
| logs.append(f"SMTP 握手失败：{repr(e)}") |
| logs.append(traceback.format_exc()) |
| return False, "\n".join(logs) |
|
|
|
|
| def test_smtp_login(smtp_server: str, smtp_port: int, mail_sender: str): |
| """ |
| 测试 SMTP 登录。 |
| 需要 MAIL_PASSWORD 在 HF Space Secrets 中配置。 |
| """ |
| if not MAIL_PASSWORD: |
| return False, ( |
| "MAIL_PASSWORD 未设置。\n\n" |
| "请在 Hugging Face Space 里配置：\n" |
| "Settings -> Variables and secrets -> New secret\n\n" |
| "MAIL_PASSWORD=你的邮箱密码或授权码" |
| ) |
|
|
| logs = [] |
|
|
| try: |
| context = ssl._create_unverified_context() |
|
|
| with smtplib.SMTP(smtp_server, smtp_port, timeout=30) as smtp: |
| code, msg = smtp.ehlo() |
| logs.append(f"EHLO 返回：code={code}, msg={msg!r}") |
|
|
| code, msg = smtp.starttls(context=context) |
| logs.append(f"STARTTLS 返回：code={code}, msg={msg!r}") |
|
|
| code, msg = smtp.ehlo() |
| logs.append(f"TLS 后 EHLO 返回：code={code}, msg={msg!r}") |
|
|
| smtp.login(mail_sender, MAIL_PASSWORD) |
| logs.append("SMTP 登录成功。") |
|
|
| return True, "\n".join(logs) |
|
|
| except Exception as e: |
| logs.append(f"SMTP 登录失败：{repr(e)}") |
| logs.append(traceback.format_exc()) |
| return False, "\n".join(logs) |
|
|
|
|
| def send_test_email( |
| smtp_server: str, |
| smtp_port: int, |
| mail_sender: str, |
| mail_to: str, |
| ): |
| """ |
| 真实发送测试邮件。 |
| """ |
| if not MAIL_PASSWORD: |
| return False, ( |
| "MAIL_PASSWORD 未设置，无法发送邮件。\n\n" |
| "请在 Hugging Face Space Secrets 中设置 MAIL_PASSWORD。" |
| ) |
|
|
| subject = "Hugging Face Space SMTP Test" |
| html_body = f""" |
| <h3>Hugging Face Space SMTP Test</h3> |
| <p>这是一封来自 Hugging Face Space 的 SMTP 测试邮件。</p> |
| <p><b>SMTP Server:</b> {smtp_server}</p> |
| <p><b>SMTP Port:</b> {smtp_port}</p> |
| <p><b>Sender:</b> {mail_sender}</p> |
| <p><b>Recipient:</b> {mail_to}</p> |
| """ |
|
|
| msg = MIMEMultipart("alternative") |
| msg["From"] = Header(mail_sender) |
| msg["To"] = Header(mail_to) |
| msg["Subject"] = Header(subject, "utf-8") |
| msg.attach(MIMEText(html_body, "html", "utf-8")) |
|
|
| logs = [] |
|
|
| try: |
| context = ssl._create_unverified_context() |
|
|
| with smtplib.SMTP(smtp_server, smtp_port, timeout=60) as smtp: |
| code, message = smtp.ehlo() |
| logs.append(f"EHLO 返回：code={code}, msg={message!r}") |
|
|
| code, message = smtp.starttls(context=context) |
| logs.append(f"STARTTLS 返回：code={code}, msg={message!r}") |
|
|
| code, message = smtp.ehlo() |
| logs.append(f"TLS 后 EHLO 返回：code={code}, msg={message!r}") |
|
|
| smtp.login(mail_sender, MAIL_PASSWORD) |
| logs.append("SMTP 登录成功。") |
|
|
| smtp.sendmail(mail_sender, [mail_to], msg.as_string()) |
| logs.append(f"测试邮件发送成功：{mail_sender} -> {mail_to}") |
|
|
| return True, "\n".join(logs) |
|
|
| except Exception as e: |
| logs.append(f"测试邮件发送失败：{repr(e)}") |
| logs.append(traceback.format_exc()) |
| return False, "\n".join(logs) |
|
|
|
|
| def run_all_tests( |
| smtp_server: str, |
| smtp_port: int, |
| mail_sender: str, |
| mail_to: str, |
| actually_send_email: bool, |
| ): |
| """ |
| 按顺序执行所有测试。 |
| """ |
| smtp_server = smtp_server.strip() |
| smtp_port = int(smtp_port) |
| mail_sender = mail_sender.strip() |
| mail_to = mail_to.strip() |
|
|
| final_logs = [] |
|
|
| final_logs.append("========== 配置 ==========") |
| final_logs.append(f"SMTP_SERVER={smtp_server}") |
| final_logs.append(f"SMTP_PORT={smtp_port}") |
| final_logs.append(f"MAIL_SENDER={mail_sender}") |
| final_logs.append(f"MAIL_TO={mail_to}") |
| final_logs.append(f"MAIL_PASSWORD_SET={bool(MAIL_PASSWORD)}") |
| final_logs.append("") |
|
|
| final_logs.append("========== 1. DNS 测试 ==========") |
| ok, log = test_dns(smtp_server, smtp_port) |
| final_logs.append(log) |
| final_logs.append("") |
|
|
| if not ok: |
| final_logs.append("结论：DNS 解析失败。") |
| final_logs.append("这不是端口问题，因为程序还没有进入连接端口阶段。") |
| return "\n".join(final_logs) |
|
|
| final_logs.append("========== 2. TCP 连接测试 ==========") |
| ok, log = test_tcp_connection(smtp_server, smtp_port) |
| final_logs.append(log) |
| final_logs.append("") |
|
|
| if not ok: |
| final_logs.append("结论：DNS 成功，但 TCP 连接失败。") |
| final_logs.append("这可能是端口、防火墙、网络策略或 SMTP 服务不允许外部访问。") |
| return "\n".join(final_logs) |
|
|
| final_logs.append("========== 3. SMTP EHLO / STARTTLS 测试 ==========") |
| ok, log = test_smtp_handshake(smtp_server, smtp_port) |
| final_logs.append(log) |
| final_logs.append("") |
|
|
| if not ok: |
| final_logs.append("结论：TCP 连接成功，但 SMTP 握手或 STARTTLS 失败。") |
| return "\n".join(final_logs) |
|
|
| final_logs.append("========== 4. SMTP 登录测试 ==========") |
| ok, log = test_smtp_login(smtp_server, smtp_port, mail_sender) |
| final_logs.append(log) |
| final_logs.append("") |
|
|
| if not ok: |
| final_logs.append("结论：SMTP 服务可访问，但登录失败。") |
| final_logs.append("请检查 MAIL_SENDER 和 MAIL_PASSWORD。") |
| return "\n".join(final_logs) |
|
|
| if actually_send_email: |
| final_logs.append("========== 5. 发送测试邮件 ==========") |
| ok, log = send_test_email( |
| smtp_server=smtp_server, |
| smtp_port=smtp_port, |
| mail_sender=mail_sender, |
| mail_to=mail_to, |
| ) |
| final_logs.append(log) |
| final_logs.append("") |
|
|
| if ok: |
| final_logs.append("最终结论：SMTP 全流程成功，邮件已发送。") |
| else: |
| final_logs.append("最终结论：登录成功，但发送邮件失败。") |
| else: |
| final_logs.append("========== 5. 发送测试邮件 ==========") |
| final_logs.append("已跳过真实发信。勾选 `Actually send test email` 后才会发送。") |
| final_logs.append("") |
| final_logs.append("最终结论：DNS、TCP、STARTTLS、登录测试均通过。") |
|
|
| return "\n".join(final_logs) |
|
|
|
|
| with gr.Blocks() as demo: |
| gr.Markdown("# Hugging Face Space SMTP 诊断工具") |
|
|
| gr.Markdown( |
| """ |
| 这个页面用于测试 Hugging Face Space 是否能访问你的 SMTP 服务器。 |
| |
| 重点判断： |
| - 如果 DNS 失败：不是端口问题。 |
| - 如果 DNS 成功但 TCP 失败：可能是端口、防火墙或网络策略问题。 |
| - 如果 TCP 成功但登录失败：可能是用户名、密码或权限问题。 |
| """ |
| ) |
|
|
| with gr.Row(): |
| smtp_server_input = gr.Textbox( |
| label="SMTP_SERVER", |
| value=DEFAULT_SMTP_SERVER, |
| ) |
| smtp_port_input = gr.Number( |
| label="SMTP_PORT", |
| value=DEFAULT_SMTP_PORT, |
| precision=0, |
| ) |
|
|
| with gr.Row(): |
| mail_sender_input = gr.Textbox( |
| label="MAIL_SENDER", |
| value=DEFAULT_MAIL_SENDER, |
| ) |
| mail_to_input = gr.Textbox( |
| label="MAIL_TO", |
| value=DEFAULT_MAIL_TO, |
| ) |
|
|
| actually_send_email = gr.Checkbox( |
| label="Actually send test email", |
| value=False, |
| ) |
|
|
| run_button = gr.Button("Run SMTP Tests") |
|
|
| output = gr.Textbox( |
| label="测试结果", |
| lines=30, |
| ) |
|
|
| run_button.click( |
| fn=run_all_tests, |
| inputs=[ |
| smtp_server_input, |
| smtp_port_input, |
| mail_sender_input, |
| mail_to_input, |
| actually_send_email, |
| ], |
| outputs=output, |
| ) |
|
|
|
|
| demo.queue() |
| demo.launch() |
