svpn

#!/usr/bin/env python3
import argparse
import hashlib
import os
import signal
from os import path

import gi
from cryptography import x509
from cryptography.hazmat.backends import default_backend

gi.require_version("Gtk", "3.0")
try:
    gi.require_version("WebKit2", "4.1")
except ValueError:
    gi.require_version("WebKit2", "4.0")

from gi.repository import Gtk, WebKit2  # noqa: E402

parser = argparse.ArgumentParser()
parser.add_argument("--host", required=True)
parser.add_argument("--print", action="store_true")
parser.add_argument(
    "--image", default="ghcr.io/fabianonunes/openfortivpn.saml.docker:0.3.0"
)
args, mass = parser.parse_known_args()


class Browser:
    window = None
    cookie = None
    terminal = None
    web_view = None
    box = None
    trusted_cert = None

    def run(self):
        Gtk.init(None)
        self.window = Gtk.Window(title="VPN", icon_name="network-vpn")
        self.window.connect("delete-event", self.quit)
        self.window.resize(500, 570)

        ctx = WebKit2.WebContext.get_default()
        cookie_storage_path = path.expanduser("~/.cache/openfortivpn.cookies")

        cookie_manager = ctx.get_cookie_manager()
        cookie_manager.set_persistent_storage(
            cookie_storage_path, WebKit2.CookiePersistentStorage.TEXT
        )

        # aceleração de hardware causa segfaults do webkit qdo rodando no wayland
        settings = WebKit2.Settings(
            hardware_acceleration_policy=WebKit2.HardwareAccelerationPolicy.NEVER
        )
        self.web_view = WebKit2.WebView(settings=settings)
        self.web_view.connect("load-failed-with-tls-errors", self.on_tls_errors)
        self.web_view.connect("load-changed", self.page_changed, cookie_manager)
        self.web_view.load_uri(f"https://{args.host}")

        self.window.add(self.web_view)

        self.window.show_all()

        return self

    def page_changed(self, webview, event, cookie_manager):
        if event != WebKit2.LoadEvent.FINISHED:
            return
        mr = webview.get_main_resource()
        cookie_manager.get_cookies(mr.get_uri(), None, self.on_receive_cookie)

    def on_receive_cookie(self, cookie_manager, result):
        if self.cookie:
            return
        cookies = cookie_manager.get_cookies_finish(result)
        cookie = next(filter(lambda c: c.get_name() == "SVPNCOOKIE", cookies), None)
        if cookie:
            self.cookie = f"{cookie.get_name()}={cookie.get_value()}"
            self.connect()

    def on_tls_errors(self, _widget, uri, certificate, errors):
        der = certificate.get_property("certificate")
        cert_x509 = x509.load_der_x509_certificate(der, default_backend())
        subject = cert_x509.subject.rfc4514_string()
        issuer = cert_x509.issuer.rfc4514_string()

        fingerprint = hashlib.sha256(der).hexdigest()
        reason = errors.first_value_nick

        if self.confirm_bad_certificate(uri, reason, subject, issuer, fingerprint):
            global mass
            mass += ["--trusted-cert", fingerprint]

            ctx = self.web_view.get_context()
            ctx.allow_tls_certificate_for_host(certificate, args.host)
            self.web_view.load_uri(uri)
            return True
        return False

    def confirm_bad_certificate(self, host, reason, subject, issuer, fingerprint):
        dialog = Gtk.MessageDialog(
            parent=None,
            buttons=Gtk.ButtonsType.YES_NO,
            image=None,
            title="Certificado inválido",
        )

        for lable in dialog.get_message_area().get_children():
            lable.set_selectable(True)

        dialog.format_secondary_markup(
            "Você está se conectando a um servidor não confiável, o que "
            "pode colocar suas credenciais em risco.\n\n"
            "<span foreground='red' weight='bold'>Gostaria de se conectar a este servidor?</span>\n\n"
            f"<b>URL</b>\n{host}\n\n"
            f"<b>Motivo</b>\n{reason}\n\n"
            f"<b>Subject</b>\n{subject}\n\n"
            f"<b>Issuer</b>\n{issuer}\n\n"
            f"<b>SHA-256 fingerprint</b>\n{fingerprint}"
        )

        response = dialog.run()
        dialog.destroy()

        return response == Gtk.ResponseType.YES

    def build_cmd(self):
        if args.print:
            print(self.cookie)
            self.quit()
            exit(0)

        priv = "env" if os.access("/var/run/docker.sock", os.W_OK | os.R_OK) else "sudo"
        cmd = (
            f"/usr/bin/{priv} docker run "
            f"--init "
            f"--rm "
            f"--network=host "
            f"--device=/dev/ppp "
            f"--cap-drop=ALL "
            f"--cap-add=DAC_OVERRIDE "
            f"--cap-add=NET_ADMIN "
            f"--volume /etc/resolv.conf:/etc/resolv.conf "
            f"--tty "
            f"--interactive "
            f"{args.image} {args.host} "
            f"--cookie {self.cookie} "
            f"{' '.join(mass)}"
        )
        return cmd.strip().split(" ")

    def quit(self, _a=None, _b=None):
        try:
            self.stop_container()
            self.window.destroy()
            Gtk.main_quit()
            while Gtk.events_pending():
                Gtk.main_iteration()
        except Exception as err:
            print(err)
            exit(1)

    def connect(self):
        scroller = self.build_vte()
        self.web_view.try_close()
        self.web_view.destroy()
        self.window.add(scroller)
        self.window.show_all()
        self.terminal.grab_focus()

    def stop_container(self, _a=None, _b=None):
        if not self.terminal:
            return
        # o openfortivpn não reage ao SIGHUP propagado pelo Vte ao sair
        try:
            self.terminal.feed_child(b"\x03")  # CTRL+C
        except:  # noqa
            # Ubuntu 18.04 ainda usa vte < 0.59.91
            self.terminal.feed_child("\x03", -1)  # noqa

    def build_vte(self):
        gi.require_version("Vte", "2.91")
        from gi.repository import Vte

        self.terminal = Vte.Terminal()
        self.terminal.spawn_async(
            Vte.PtyFlags.DEFAULT,  # pty_flags
            os.environ["HOME"],  # working_directory
            self.build_cmd(),  # argv
            [],  # envv
            0,  # spawn_flags: GLib.SpawnFlags.DEFAULT
            None,  # child_setup
            None,  # child_setup_data - não documentado
            -1,  # timeout
        )

        scroller = Gtk.ScrolledWindow()
        scroller.add(self.terminal)
        return scroller


if __name__ == "__main__":
    b = Browser().run()
    signal.signal(signal.SIGINT, b.quit)
    Gtk.main()