From d26f2e40d49e8ce1f01d4f1db4a54c2c6fb91964 Mon Sep 17 00:00:00 2001 From: Ivan Date: Wed, 8 Apr 2026 16:17:00 -0500 Subject: [PATCH] feat(meshchat): add utility modules for environment variable parsing, file path resolution, and self-signed SSL certificate generation --- meshchatx/src/env_utils.py | 10 +++++ meshchatx/src/path_utils.py | 71 +++++++++++++++++++++++++++++ meshchatx/src/ssl_self_signed.py | 76 ++++++++++++++++++++++++++++++++ 3 files changed, 157 insertions(+) create mode 100644 meshchatx/src/env_utils.py create mode 100644 meshchatx/src/path_utils.py create mode 100644 meshchatx/src/ssl_self_signed.py diff --git a/meshchatx/src/env_utils.py b/meshchatx/src/env_utils.py new file mode 100644 index 0000000..66fe859 --- /dev/null +++ b/meshchatx/src/env_utils.py @@ -0,0 +1,10 @@ +"""Environment variable parsing helpers.""" + +import os + + +def env_bool(env_name, default=False): + val = os.environ.get(env_name) + if val is None: + return default + return val.lower() in ("true", "1", "yes", "on") diff --git a/meshchatx/src/path_utils.py b/meshchatx/src/path_utils.py new file mode 100644 index 0000000..59a8c66 --- /dev/null +++ b/meshchatx/src/path_utils.py @@ -0,0 +1,71 @@ +"""Filesystem and HTTP client helpers used at startup and in the web layer.""" + +import os +import sys +import tempfile +from aiohttp import web + + +def resolve_log_dir(): + """Choose a writable log directory across container, desktop, and Windows.""" + env_dir = os.environ.get("MESHCHAT_LOG_DIR") + candidates = [] + if env_dir: + candidates.append(env_dir) + + candidates.append("/config/logs") + + if os.name == "nt": + appdata = os.environ.get("LOCALAPPDATA") or os.environ.get("APPDATA") + if appdata: + candidates.append(os.path.join(appdata, "MeshChatX", "logs")) + + home_dir = os.path.expanduser("~") + candidates.append(os.path.join(home_dir, ".reticulum-meshchatx", "logs")) + candidates.append(os.path.join(tempfile.gettempdir(), "meshchatx", "logs")) + + for path in candidates: + if not path: + continue + try: + os.makedirs(path, exist_ok=True) + return path + except PermissionError: + continue + except OSError: + continue + + return None + + +def request_client_ip(request: web.Request) -> str: + xff = request.headers.get("X-Forwarded-For") + if xff: + return xff.split(",")[0].strip() + if request.remote: + return request.remote + return "" + + +def get_file_path(filename): + # NOTE: this is required to be able to pack our app with cxfreeze as an exe, otherwise it can't access bundled assets + # this returns a file path based on if we are running meshchat.py directly, or if we have packed it as an exe with cxfreeze + # https://cx-freeze.readthedocs.io/en/latest/faq.html#using-data-files + # bearer:disable python_lang_path_traversal + filename = filename.rstrip("/\\") + + if getattr(sys, "frozen", False): + datadir = os.path.dirname(sys.executable) + return os.path.join(datadir, filename) + + package_dir = os.path.dirname(os.path.dirname(__file__)) + package_path = os.path.join(package_dir, filename) + if os.path.exists(package_path): + return package_path + + repo_root = os.path.dirname(package_dir) + repo_path = os.path.join(repo_root, filename) + if os.path.exists(repo_path): + return repo_path + + return package_path diff --git a/meshchatx/src/ssl_self_signed.py b/meshchatx/src/ssl_self_signed.py new file mode 100644 index 0000000..8dbb197 --- /dev/null +++ b/meshchatx/src/ssl_self_signed.py @@ -0,0 +1,76 @@ +"""Self-signed TLS certificate generation for local HTTPS.""" + +import ipaddress +import os +from datetime import UTC, datetime, timedelta + +from cryptography import x509 +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import NameOID + + +def generate_ssl_certificate(cert_path: str, key_path: str): + """Generate a self-signed SSL certificate for local HTTPS. + + Args: + cert_path: Path where the certificate will be saved + key_path: Path where the private key will be saved + + """ + if os.path.exists(cert_path) and os.path.exists(key_path): + return + + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + backend=default_backend(), + ) + + subject = issuer = x509.Name( + [ + x509.NameAttribute(NameOID.COUNTRY_NAME, "US"), + x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "Local"), + x509.NameAttribute(NameOID.LOCALITY_NAME, "Local"), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Reticulum MeshChatX"), + x509.NameAttribute(NameOID.COMMON_NAME, "localhost"), + ], + ) + + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(private_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(datetime.now(UTC)) + .not_valid_after(datetime.now(UTC) + timedelta(days=365)) + .add_extension( + x509.SubjectAlternativeName( + [ + x509.DNSName("localhost"), + x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")), + x509.IPAddress(ipaddress.IPv6Address("::1")), + ], + ), + critical=False, + ) + .sign(private_key, hashes.SHA256(), default_backend()) + ) + + cert_dir = os.path.dirname(cert_path) + if cert_dir: + os.makedirs(cert_dir, exist_ok=True) + + with open(cert_path, "wb") as f: + f.write(cert.public_bytes(serialization.Encoding.PEM)) + + with open(key_path, "wb") as f: + f.write( + private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ), + )