| |
| """Run the existing log watcher with its SMTP connection tunneled via HTTP proxy.""" |
|
|
| from __future__ import annotations |
|
|
| import os |
| import runpy |
| import smtplib |
| import sys |
| from pathlib import Path |
| from urllib.parse import urlsplit |
|
|
| import socks |
|
|
|
|
| def main() -> None: |
| proxy_value = ( |
| os.environ.get("HTTPS_PROXY") |
| or os.environ.get("https_proxy") |
| or os.environ.get("HTTP_PROXY") |
| or os.environ.get("http_proxy") |
| ) |
| if not proxy_value: |
| raise SystemExit("HTTP(S) proxy environment variable is required") |
|
|
| proxy = urlsplit(proxy_value) |
| if not proxy.hostname or not proxy.port: |
| raise SystemExit("Invalid HTTP proxy URL") |
|
|
| def get_proxy_socket( |
| _smtp: smtplib.SMTP, |
| host: str, |
| port: int, |
| timeout: float | None, |
| ): |
| return socks.create_connection( |
| (host, port), |
| timeout=timeout, |
| proxy_type=socks.HTTP, |
| proxy_addr=proxy.hostname, |
| proxy_port=proxy.port, |
| proxy_username=proxy.username, |
| proxy_password=proxy.password, |
| proxy_rdns=True, |
| ) |
|
|
| smtplib.SMTP._get_socket = get_proxy_socket |
| repo_root = Path(__file__).resolve().parents[1] |
| watcher = repo_root / "log_watcher" / "watch_log_email.py" |
| sys.argv[0] = str(watcher) |
| runpy.run_path(str(watcher), run_name="__main__") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
