#!/usr/bin/python3 -su
# pylint: disable=invalid-name,broad-exception-caught,global-statement

"""
Copyright (C) 2014 - 2015 Jason Mehring <nrgaway@gmail.com>
License: GPL-2+
Authors: Jason Mehring

  This program is free software; you can redistribute it and/or
  modify it under the terms of the GNU General Public License
  as published by the Free Software Foundation; either version 2
  of the License, or (at your option) any later version.

  This program is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  GNU General Public License for more details.

  You should have received a copy of the GNU General Public License
  along with this program.  If not, see <https://www.gnu.org/licenses/>.

replace-ips - Search and replace IP addresses in specified files.

All Whonix configuration files provided are searched for the last known
occurrence of an IP address that was used and replaced with the current IP
address provided by Qubes.

Initially, the known defaults in Whonix configuration files IP addresses are:
- Whonix-Gateway    : eth1: 10.152.152.10 | fd19:c33d:88bc::10
- Whonix-Workstation: eth0: 10.152.152.11 | fd19:c33d:88bc::11

Tasks of this script:

- Whonix-Gateway (sys-whonix):
  - Replace file '/usr/share/tor/tor-service-defaults-torrc.anondist.base'
    default 'eth1' IP '10.152.152.10' with qubes_ipv4_qubesdb_qubes_netvm_gateway
    (qubesdb-read /qubes-netvm-gateway).
    Purpose: Tor can bind to 'eth1'.
  - Replace file '/usr/local/etc/torrc.d/50_user.conf'. Same as above.
    Purpose: Tor onion service can bind to 'eth1'.

- Whonix-Workstation (anon-whonix):
  - Replace file '/etc/resolv.conf' default IP 'eth0' '10.152.152.10' with
    qubes_ipv4_qubesdb_qubes_gateway (qubesdb-read /qubes-gateway).
    Purpose: faster DNS resolution.
  - Replace Whonix-Workstation's own local IP '10.152.152.11' with
    qubes_ipv4_qubesdb_qubes_ip (qubesdb-read /qubes-ip).
    Purpose: onion services such as a HTTP server binding to eth0 IP address.

They are also checked each time this module is run in case the configuration
files were modified due to a system update.

Qubes feature request: optional static IP addresses
https://github.com/QubesOS/qubes-issues/issues/1477
"""

import os
import sys
import re
import subprocess
import ipaddress
import inspect
from typing import Pattern, Any
from types import FrameType

IP_WAS_REPLACED: bool = False

WHONIX_GATEWAY_IP4_ETH1_SAVE_FILE: str = "/var/cache/qubes-whonix/whonix-ip-gateway"
WHONIX_GATEWAY_IP6_ETH1_SAVE_FILE: str = "/var/cache/qubes-whonix/whonix-ip6-gateway"
WHONIX_WORKSTATION_IP4_ETH0_SAVE_FILE: str = "/var/cache/qubes-whonix/whonix-ip-local"
WHONIX_WORKSTATION_IP6_ETH0_SAVE_FILE: str = "/var/cache/qubes-whonix/whonix-ip6-local"

SEARCH_AND_REPLACE_FILE_LIST: list[str]
# This is a list of all Whonix files that contain IP addresses that will
# be searched and replaced with the currently assigned IP address
if os.path.exists("/usr/share/anon-gw-base-files/gateway"):
    SEARCH_AND_REPLACE_FILE_LIST = [
        "/usr/share/whonix-gw-network-conf/network_internal_ip.txt",
        "/etc/resolv.conf",
        "/etc/resolv.conf.whonix",
        "/etc/resolv.conf.anondist",
        "/etc/tor/torrc",
        "/usr/local/etc/torrc.d/50_user.conf",
        "/usr/share/tor/tor-service-defaults-torrc.anondist.base",
        "/usr/share/tor/tor-service-defaults-torrc.anondist",
    ]
elif os.path.exists("/usr/share/anon-ws-base-files/workstation"):
    SEARCH_AND_REPLACE_FILE_LIST = [
        "/etc/resolv.conf",
        "/etc/resolv.conf.whonix",
        "/etc/resolv.conf.anondist",
        "/home/user/.xchat2/xchat.conf",
        "/home/user/.config/hexchat/hexchat.conf",
        "/usr/libexec/helper-scripts/leak-tests/simple_ping.py",
        "/usr/share/anon-apps-config/kioslaverc",
    ]
else:
    print("ERROR: Neither Whonix-Gateway nor Whonix-Workstation detected. Please report this bug.")
    sys.exit(1)


whonix_gateway_ipv4_eth1_default: str = "10.152.152.10"
whonix_gateway_ipv6_eth1_default: str = "fd19:c33d:88bc::10"
whonix_workstation_ipv4_eth0_default: str = "10.152.152.11"
whonix_workstation_ipv6_eth0_default: str = "fd19:c33d:88bc::11"


def var_name(obj: Any, up: int = 1) -> str | None:
    """
    Best-effort: return the name of `obj` in a frame `up` levels above this one.

    up = 1 -> immediate caller
    up = 2 -> caller's caller, etc.
    """
    frame_arr: list[FrameType | None] = [inspect.currentframe()]
    try:
        # walk `up` frames back starting from this helper's frame
        for _ in range(up):
            if frame_arr[-1] is None:
                return None
            frame_arr.append(frame_arr[-1].f_back)
        if frame_arr[-1] is None:
            return None

        for name, value in frame_arr[-1].f_locals.items():
            if value is obj:
                return name
        return None
    finally:
        # avoid reference cycles
        for frame in reversed(frame_arr):
            del frame


def whonix_mode() -> str:
    """Determine Whonix mode.

    Can be either 'gateway', 'workstation', 'template', or 'unknown'.
    """
    mode: str = "unknown"
    if os.path.exists("/run/qubes/this-is-templatevm"):
        mode = "template"
    elif os.path.exists("/usr/share/anon-gw-base-files/gateway"):
        mode = "gateway"
    elif os.path.exists("/usr/share/anon-ws-base-files/workstation"):
        mode = "workstation"

    return mode


def ip_line_fixup(line: str, last_ip: str, current_ip: str) -> str:
    """Does low-level IP address replacement via regex matches.
    """

    ## NOTE: This looks horribly wasteful on the surface, but it's not as bad
    ## as it looks, as Python caches regexes, and line-by-line processing
    ## skipping irrelevant lines gives the regexes a lot less to do. The only
    ## really wasteful thing here is the repeated escaping of last_ip and
    ## splitting of current_ip, both of which are probably pretty fast
    ## operations. Do NOT attempt to optimize this by using regex matches that
    ## apply to entire files at once, this actually made the script slower!

    comment_regex: Pattern[str] = re.compile(r"\s*#")
    if line == "":
        return line
    if comment_regex.match(line):
        return line

    if last_ip in line:
        line = re.sub(
            rf"(?m){re.escape(last_ip)}(?=\D|$)",
            current_ip,
            line,
        )

    if ":" in last_ip:
        line = re.sub(
            rf"(?m){re.escape(last_ip.rsplit(":", 1)[0])}:0"
            r"(?=\D|$)",
            current_ip.rsplit(":", 1)[0] + ":0",
            line,
        )
    else:
        line = re.sub(
            rf"(?m){re.escape(last_ip.rsplit(".", 1)[0])}\.0"
            r"(?=\D|$)",
            current_ip.rsplit(".", 1)[0] + ".0",
            line,
        )

    return line


# pylint: disable=too-many-locals,too-many-branches
def replace_ip(
    ips: list[str],
    current_ip: str,
    files: list[str],
    ip_file: str
) -> bool:
    """Searches and replaces IP addresses in the provided files.

    ips:
        List of IP addresses to replace.

    current_ip:
        IP replacement address.

    files:
        List of files to search. The list must contain full pathnames.

    ip_file:
        Full path to the filename used to store the last known value of the IP
        address. The `current_ip` is stored in this file and used the next
        time this module is executed.
    """
    global IP_WAS_REPLACED

    replaced: bool = False

    ips_varname = var_name(ips, up=2)
    ip_file_varname = var_name(ip_file, up=3)
    current_ip_varname = var_name(current_ip, up=2)

    for ip_item in ips:
        try:
            ipaddress.ip_address(ip_item)
            # print(f"INFO: ip_item variable valid: {ip_item}")
        except ValueError:
            print(f"ERROR: ip_item variable invalid: {ip_item}")
            return replaced

    try:
        ipaddress.ip_address(current_ip)
        # print(f"INFO: current_ip variable valid: '{current_ip_varname}' '{current_ip}'")
    except ValueError:
        print(f"ERROR: current_ip variable invalid: '{current_ip_varname}' '{current_ip}'")
        return replaced

    protocol: str
    if ":" in current_ip:
        protocol = "IPv6"
    else:
        protocol = "IPv4"

    for filename in files:
        if not os.path.exists(filename):
            continue
        try:
            with open(
                filename, "r", encoding="utf-8", errors="surrogateescape"
            ) as infile:
                text: str = infile.read()
        except IOError:
            print(
                f"ERROR: {protocol}: file existing but failed to open for "
                f"reading: '{filename}'"
            )
            continue

        line_list: list[str] = text.split("\n")
        for idx, line in enumerate(line_list):
            for last_ip in ips:
                line = ip_line_fixup(line, last_ip, current_ip)
                line_list[idx] = line

        replaced_text: str = "\n".join(line_list)

        if text != replaced_text:
            try:
                with open(filename, "w", encoding="utf-8") as outfile:
                    outfile.write(replaced_text)
                replaced = True
                print(
                    f"INFO: '{protocol}': filename from filelist updated  : "
                    f"'{filename}' '{ips_varname}' '{current_ip_varname}' '{current_ip}'"
                )
            except IOError:
                print(
                    f"ERROR: '{protocol}': filename from filelist existing "
                    f"but failed to open filename for writing: '{filename}'"
                )
                continue
                # return False
        # else:
            # print(
            #     f"INFO: '{protocol}': filename from filelist unchanged: "
            #     f"'{filename}'"
            # )
    if replaced:
        try:
            with open(ip_file, "w", encoding="utf-8") as outfile:
                outfile.write(current_ip)
            replaced = True
            print(
                f"INFO: '{protocol}': ip_file updated: "
                f"'{ip_file_varname}' '{ip_file}' '{current_ip_varname}' '{current_ip}'"
            )
            IP_WAS_REPLACED = True
        except IOError:
            print(
                f"ERROR: '{protocol}': writing to ip_file failed: "
                f"'{ip_file_varname}' '{current_ip_varname}' '{current_ip}'"
            )
            return False

    return replaced


def get_ip_address(filename: str, default: str = "") -> str:
    """Retrieve an IP address from a file."""
    filename_varname = var_name(filename, up=3)

    if not os.path.exists(filename):
        print(
            f"INFO: IP filename does not exist (returning default): "
            f"'{filename_varname}' '{filename}'"
        )
        return default

    try:
        with open(
            filename, "r", encoding="utf-8", errors="surrogateescape"
        ) as infile:
            ip_txt: str = infile.read().strip()
    except (OSError, IOError):
        print(
            f"INFO: IP filename exists but opening failed: "
            f"'{filename_varname}' '{filename}'"
        )
        return default

    try:
        ipaddress.ip_address(ip_txt)
        print(
            f"INFO: IP filename reading succeeded and valid IP: "
            f"'{filename_varname}' '{filename}'"
        )
        return ip_txt
    except ValueError:
        print(
            f"INFO: IP filename reading succeeded but not a valid IP "
            f"(returning default): '{filename_varname}' '{filename}'"
        )
        return default


def maybe_reload_tor() -> None:
    """Reload Tor's configuration files if Tor is currently active and not
    disabled.
    """
    try:
        if subprocess.check_output(["systemctl", "is-active", "tor@default"]):
            print("INFO: executing: systemctl restart tor@default")
            # Restarting instead of reloading due to upstream Tor bug
            # https://gitlab.torproject.org/tpo/core/tor/-/issues/16161
            subprocess.call(["systemctl", "restart", "tor@default"])
    except subprocess.CalledProcessError:
        print(
            "INFO: Systemd unit tor@default is not running, therefore not "
            "restarting."
        )


# pylint: disable=too-many-branches,too-many-statements
def main() -> None:
    """Main function.
    """
    global IP_WAS_REPLACED
    IP_WAS_REPLACED = False

    mode = whonix_mode()

    if mode == "template":
        print("INFO: Not needed inside Template, skipping, ok.")
        return

    if mode == "unknown":
        print(
            "ERROR: Neither Whonix-Gateway, nor Whonix-Workstation, nor "
            "Template detected. Please report this bug."
        )
        sys.exit(1)

    ## IP HARDCODED, but this does not matter for Non-Qubes-Whonix. This script
    ## is currently only used in Qubes-Whonix.
    whonix_gateway_ipv4_eth1_last: str = get_ip_address(
        WHONIX_GATEWAY_IP4_ETH1_SAVE_FILE, whonix_gateway_ipv4_eth1_default
    )
    whonix_gateway_ipv6_eth1_last: str = get_ip_address(
        WHONIX_GATEWAY_IP6_ETH1_SAVE_FILE, whonix_gateway_ipv6_eth1_default
    )
    whonix_workstation_ipv4_eth0_last: str = get_ip_address(
        WHONIX_WORKSTATION_IP4_ETH0_SAVE_FILE, whonix_workstation_ipv4_eth0_default
    )
    whonix_workstation_ipv6_eth0_last: str = get_ip_address(
        WHONIX_WORKSTATION_IP6_ETH0_SAVE_FILE, whonix_workstation_ipv6_eth0_default
    )

    assert whonix_gateway_ipv4_eth1_last != ""
    assert whonix_gateway_ipv6_eth1_last != ""
    assert whonix_workstation_ipv4_eth0_last != ""
    assert whonix_workstation_ipv6_eth0_last != ""

    print(f"INFO: whonix_gateway_ipv4_eth1_last    : {whonix_gateway_ipv4_eth1_last}")
    print(f"INFO: whonix_gateway_ipv6_eth1_last    : {whonix_gateway_ipv6_eth1_last}")
    print(f"INFO: whonix_workstation_ipv4_eth0_last: {whonix_workstation_ipv4_eth0_last}")
    print(f"INFO: whonix_workstation_ipv6_eth0_last: {whonix_workstation_ipv6_eth0_last}")

    qubes_ipv4_qubesdb_qubes_ip: str | None = None
    qubes_ipv4_qubesdb_qubes_netvm_gateway: str | None = None
    qubes_ipv4_qubesdb_qubes_gateway: str | None = None
    qubes_ipv6_qubesdb_qubes_ip6: str | None = None
    qubes_ipv6_qubesdb_qubes_netvm_gateway6: str | None = None
    qubes_ipv6_qubesdb_qubes_gateway6: str | None = None

    if not os.path.isdir("/var/cache/qubes-whonix"):
        try:
            os.makedirs("/var/cache/qubes-whonix")
        except Exception:
            print("ERROR: could not create folder '/var/cache/qubes-whonix'.")

    if mode == "gateway":
        try:
            qubes_ipv4_qubesdb_qubes_netvm_gateway = (
                subprocess.check_output(
                    ["qubesdb-read", "/qubes-netvm-gateway"]
                )
                .decode()
                .rstrip()
            )
        except (OSError, subprocess.CalledProcessError):
            print("WARNING: 'qubesdb-read /qubes-netvm-gateway' failed!")
        print(
            "INFO: qubes_ipv4_qubesdb_qubes_netvm_gateway : "
            f"{qubes_ipv4_qubesdb_qubes_netvm_gateway}"
        )

        try:
            qubes_ipv6_qubesdb_qubes_netvm_gateway6 = (
                subprocess.check_output(
                    ["qubesdb-read", "/qubes-netvm-gateway6"]
                )
                .decode()
                .rstrip()
            )
        except (OSError, subprocess.CalledProcessError):
            print("WARNING: 'qubesdb-read /qubes-netvm-gateway6' failed!")
        print(
            "INFO: qubes_ipv6_qubesdb_qubes_netvm_gateway6: "
            f"{qubes_ipv6_qubesdb_qubes_netvm_gateway6}"
        )

        if qubes_ipv4_qubesdb_qubes_netvm_gateway is not None:
            gateway_ipv4_eth1_ips_to_replace: list[str] = [
                whonix_gateway_ipv4_eth1_last,
                whonix_gateway_ipv4_eth1_default,
            ]
            replace_ip(
                gateway_ipv4_eth1_ips_to_replace,
                qubes_ipv4_qubesdb_qubes_netvm_gateway,
                SEARCH_AND_REPLACE_FILE_LIST,
                WHONIX_GATEWAY_IP4_ETH1_SAVE_FILE,
            )

        if qubes_ipv6_qubesdb_qubes_netvm_gateway6 is not None:
            gateway_ipv6_eth1_ips_to_replace: list[str] = [
                whonix_gateway_ipv6_eth1_last,
                whonix_gateway_ipv6_eth1_default,
            ]
            replace_ip(
                gateway_ipv6_eth1_ips_to_replace,
                qubes_ipv6_qubesdb_qubes_netvm_gateway6,
                SEARCH_AND_REPLACE_FILE_LIST,
                WHONIX_GATEWAY_IP6_ETH1_SAVE_FILE,
            )

        if IP_WAS_REPLACED:
            maybe_reload_tor()

    if mode == "workstation":
        try:
            qubes_ipv4_qubesdb_qubes_ip = (
                subprocess.check_output(["qubesdb-read", "/qubes-ip"])
                .decode()
                .rstrip()
            )
            qubes_ipv4_qubesdb_qubes_gateway = (
                subprocess.check_output(["qubesdb-read", "/qubes-gateway"])
                .decode()
                .rstrip()
            )
        except (OSError, subprocess.CalledProcessError):
            print(
                "WARNING: 'qubesdb-read /qubes-ip' or 'qubesdb-read "
                "/qubes-gateway' failed!"
            )
        print(f"INFO: qubes_ipv4_qubesdb_qubes_ip      : {qubes_ipv4_qubesdb_qubes_ip}")
        print(
            f"INFO: qubes_ipv4_qubesdb_qubes_gateway : "
            f"{qubes_ipv4_qubesdb_qubes_gateway}"
        )

        try:
            qubes_ipv6_qubesdb_qubes_ip6 = (
                subprocess.check_output(["qubesdb-read", "/qubes-ip6"])
                .decode()
                .rstrip()
            )
            qubes_ipv6_qubesdb_qubes_gateway6 = (
                subprocess.check_output(["qubesdb-read", "/qubes-gateway6"])
                .decode()
                .rstrip()
            )
        except (OSError, subprocess.CalledProcessError):
            print(
                "WARNING: 'qubesdb-read /qubes-ip6' or 'qubesdb-read "
                "/qubes-gateway6' failed!"
            )
        print(f"INFO: qubes_ipv6_qubesdb_qubes_ip6     : {qubes_ipv6_qubesdb_qubes_ip6}")
        print(
            f"INFO: qubes_ipv6_qubesdb_qubes_gateway6: "
            f"{qubes_ipv6_qubesdb_qubes_gateway6}"
        )

        if None not in (
            qubes_ipv4_qubesdb_qubes_ip,
            qubes_ipv4_qubesdb_qubes_gateway,
        ):
            assert qubes_ipv4_qubesdb_qubes_ip is not None
            assert qubes_ipv4_qubesdb_qubes_gateway is not None

            # Workstation's own IPv4 (10.152.152.11 -> /qubes-ip)
            workstation_ipv4_eth0_ips_to_replace: list[str] = [
                whonix_workstation_ipv4_eth0_last,
                whonix_workstation_ipv4_eth0_default,
            ]
            replace_ip(
                workstation_ipv4_eth0_ips_to_replace,
                qubes_ipv4_qubesdb_qubes_ip,
                SEARCH_AND_REPLACE_FILE_LIST,
                WHONIX_WORKSTATION_IP4_ETH0_SAVE_FILE,
            )

            # Gateway IPv4 as seen on workstation (10.152.152.10 -> /qubes-gateway)
            gateway_ipv4_eth1_ips_for_workstation: list[str] = [
                whonix_gateway_ipv4_eth1_last,
                whonix_gateway_ipv4_eth1_default,
            ]
            replace_ip(
                gateway_ipv4_eth1_ips_for_workstation,
                qubes_ipv4_qubesdb_qubes_gateway,
                SEARCH_AND_REPLACE_FILE_LIST,
                WHONIX_GATEWAY_IP4_ETH1_SAVE_FILE,
            )

        if None not in (
            qubes_ipv6_qubesdb_qubes_ip6,
            qubes_ipv6_qubesdb_qubes_gateway6,
        ):
            assert qubes_ipv6_qubesdb_qubes_ip6 is not None
            assert qubes_ipv6_qubesdb_qubes_gateway6 is not None

            # Workstation's own IPv6
            workstation_ipv6_eth0_ips_to_replace: list[str] = [
                whonix_workstation_ipv6_eth0_last,
                whonix_workstation_ipv6_eth0_default,
            ]
            replace_ip(
                workstation_ipv6_eth0_ips_to_replace,
                qubes_ipv6_qubesdb_qubes_ip6,
                SEARCH_AND_REPLACE_FILE_LIST,
                WHONIX_WORKSTATION_IP6_ETH0_SAVE_FILE,
            )

            # Gateway IPv6 as seen on workstation
            gateway_ipv6_eth1_ips_for_workstation: list[str] = [
                whonix_gateway_ipv6_eth1_last,
                whonix_gateway_ipv6_eth1_default,
            ]
            replace_ip(
                gateway_ipv6_eth1_ips_for_workstation,
                qubes_ipv6_qubesdb_qubes_gateway6,
                SEARCH_AND_REPLACE_FILE_LIST,
                WHONIX_GATEWAY_IP6_ETH1_SAVE_FILE,
            )


if __name__ == "__main__":
    print("/usr/lib/qubes-whonix/replace-ips INFO: START")
    main()
    print("/usr/lib/qubes-whonix/replace-ips INFO: END")
