WebAisMap/ssl_utils.py

import os
import ssl
import socket
import datetime


def _get_local_ips():
    """Возвращает список локальных IP-адресов машины."""
    ips = []
    try:
        for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
            ip = info[4][0]
            if ip not in ips:
                ips.append(ip)
    except Exception:
        pass
    return ips


def get_ssl_context():
    """
    Создаёт собственный мини-CA и подписывает серверный сертификат.
    CA-сертификат можно установить на телефон один раз — и предупреждения исчезнут.
    Файлы: ca.pem (корневой, для установки на клиенты), cert.pem + key.pem (сервер).
    """
    cert_dir = os.path.dirname(os.path.abspath(__file__))
    ca_cert_file = os.path.join(cert_dir, "ca.pem")
    ca_key_file = os.path.join(cert_dir, "ca_key.pem")
    cert_file = os.path.join(cert_dir, "cert.pem")
    key_file = os.path.join(cert_dir, "key.pem")

    need_regen = not all(os.path.exists(f) for f in [ca_cert_file, ca_key_file, cert_file, key_file])

    if need_regen:
        try:
            from cryptography import x509
            from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID
            from cryptography.hazmat.primitives import hashes, serialization
            from cryptography.hazmat.primitives.asymmetric import rsa
            import ipaddress
        except ImportError:
            print("[SSL] Библиотека cryptography не установлена.")
            print("[SSL] Установите:  pip install cryptography")
            print("[SSL] Без HTTPS GPS телефона работать не будет.")
            return None

        print("[SSL] Генерация корневого CA и серверного сертификата...")

        # === 1. Корневой CA ===
        ca_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
        ca_name = x509.Name([
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, "AIS Map CA"),
            x509.NameAttribute(NameOID.COMMON_NAME, "AIS Map Root CA"),
        ])
        ca_cert = (
            x509.CertificateBuilder()
            .subject_name(ca_name)
            .issuer_name(ca_name)
            .public_key(ca_key.public_key())
            .serial_number(x509.random_serial_number())
            .not_valid_before(datetime.datetime.utcnow())
            .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=3650))
            .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True)
            .add_extension(
                x509.KeyUsage(
                    digital_signature=True, key_cert_sign=True, crl_sign=True,
                    content_commitment=False, key_encipherment=False,
                    data_encipherment=False, key_agreement=False,
                    encipher_only=False, decipher_only=False,
                ),
                critical=True,
            )
            .sign(ca_key, hashes.SHA256())
        )

        with open(ca_cert_file, "wb") as f:
            f.write(ca_cert.public_bytes(serialization.Encoding.PEM))
        with open(ca_key_file, "wb") as f:
            f.write(ca_key.private_bytes(
                serialization.Encoding.PEM,
                serialization.PrivateFormat.TraditionalOpenSSL,
                serialization.NoEncryption(),
            ))
        print(f"[SSL] CA сертификат: {ca_cert_file}")

        # === 2. Серверный сертификат, подписанный CA ===
        srv_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
        srv_name = x509.Name([
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, "AIS Map"),
            x509.NameAttribute(NameOID.COMMON_NAME, "aismap.local"),
        ])

        san_entries = [x509.DNSName("aismap.local"), x509.DNSName("localhost")]
        for iface_ip in _get_local_ips():
            try:
                san_entries.append(x509.IPAddress(ipaddress.IPv4Address(iface_ip)))
            except Exception:
                pass
        san_entries.append(x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")))

        srv_cert = (
            x509.CertificateBuilder()
            .subject_name(srv_name)
            .issuer_name(ca_name)
            .public_key(srv_key.public_key())
            .serial_number(x509.random_serial_number())
            .not_valid_before(datetime.datetime.utcnow())
            .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=3650))
            .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
            .add_extension(x509.SubjectAlternativeName(san_entries), critical=False)
            .add_extension(
                x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]),
                critical=False,
            )
            .sign(ca_key, hashes.SHA256())
        )

        with open(cert_file, "wb") as f:
            f.write(srv_cert.public_bytes(serialization.Encoding.PEM))
        with open(key_file, "wb") as f:
            f.write(srv_key.private_bytes(
                serialization.Encoding.PEM,
                serialization.PrivateFormat.TraditionalOpenSSL,
                serialization.NoEncryption(),
            ))
        print(f"[SSL] Серверный сертификат: {cert_file}")

    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain(cert_file, key_file)
    return ctx


def run_http_redirect(https_port):
    """Простой HTTP-сервер на порту 80, который редиректит на HTTPS."""
    from flask import Flask, redirect, request as flask_request
    redir_app = Flask("redirect")

    @redir_app.route("/", defaults={"path": ""})
    @redir_app.route("/<path:path>")
    def redir(path):
        host = flask_request.host.split(":")[0]
        port_suffix = f":{https_port}" if https_port != 443 else ""
        return redirect(f"https://{host}{port_suffix}/{path}", code=301)

    try:
        redir_app.run(host="0.0.0.0", port=80, threaded=True)
    except Exception as e:
        print(f"[HTTP] Не удалось запустить редирект на порту 80: {e}")