Create: MVP app

.gitignore

@@ -160,3 +160,6 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
 
+# Specific
+**/etc/*
+**/result

CHANGELOG.md

@@ -0,0 +1 @@
+# Changelog

flake.lock

@@ -0,0 +1,27 @@
+{
+  "nodes": {
+    "nixpkgs": {
+      "locked": {
+        "lastModified": 1689935543,
+        "narHash": "sha256-6GQ9ib4dA/r1leC5VUpsBo0BmDvNxLjKrX1iyL+h8mc=",
+        "owner": "nixos",
+        "repo": "nixpkgs",
+        "rev": "e43e2448161c0a2c4928abec4e16eae1516571bc",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nixos",
+        "ref": "nixpkgs-unstable",
+        "repo": "nixpkgs",
+        "type": "github"
+      }
+    },
+    "root": {
+      "inputs": {
+        "nixpkgs": "nixpkgs"
+      }
+    }
+  },
+  "root": "root",
+  "version": 7
+}

flake.nix

@@ -0,0 +1,30 @@
+{
+  description = "wg-quick wrapper for creating interface config and up/down wg";
+  inputs = {
+    nixpkgs.url = "github:nixos/nixpkgs/nixpkgs-unstable";
+  };
+
+  outputs = { self, nixpkgs }:
+  let
+    system = "x86_64-linux";
+    pkgs = nixpkgs.legacyPackages.${system};
+  in{
+    packages.${system}.default = pkgs.poetry2nix.mkPoetryApplication {
+      projectDir = self;
+    };
+
+    devShells.${system}.default = pkgs.mkShellNoCC {
+      shellHook = "echo Welcome to your Nix-powered development environment!"; 
+      WG_BOOTSTRAP = "development";
+      packages = with pkgs; [
+        (poetry2nix.mkPoetryEnv { projectDir = self; })
+        wireguard-tools
+      ];
+    };
+
+    apps.${system}.default = {
+      program = "${self.packages.${system}.default}/bin/wg-start";
+      type = "app";
+    };
+  };
+}

poetry.lock

@@ -0,0 +1,55 @@
+# This file is automatically @generated by Poetry 1.5.1 and should not be changed by hand.
+
+[[package]]
+name = "flake8"
+version = "6.0.0"
+description = "the modular source code checker: pep8 pyflakes and co"
+optional = false
+python-versions = ">=3.8.1"
+files = [
+    {file = "flake8-6.0.0-py2.py3-none-any.whl", hash = "sha256:3833794e27ff64ea4e9cf5d410082a8b97ff1a06c16aa3d2027339cd0f1195c7"},
+    {file = "flake8-6.0.0.tar.gz", hash = "sha256:c61007e76655af75e6785a931f452915b371dc48f56efd765247c8fe68f2b181"},
+]
+
+[package.dependencies]
+mccabe = ">=0.7.0,<0.8.0"
+pycodestyle = ">=2.10.0,<2.11.0"
+pyflakes = ">=3.0.0,<3.1.0"
+
+[[package]]
+name = "mccabe"
+version = "0.7.0"
+description = "McCabe checker, plugin for flake8"
+optional = false
+python-versions = ">=3.6"
+files = [
+    {file = "mccabe-0.7.0-py2.py3-none-any.whl", hash = "sha256:6c2d30ab6be0e4a46919781807b4f0d834ebdd6c6e3dca0bda5a15f863427b6e"},
+    {file = "mccabe-0.7.0.tar.gz", hash = "sha256:348e0240c33b60bbdf4e523192ef919f28cb2c3d7d5c7794f74009290f236325"},
+]
+
+[[package]]
+name = "pycodestyle"
+version = "2.10.0"
+description = "Python style guide checker"
+optional = false
+python-versions = ">=3.6"
+files = [
+    {file = "pycodestyle-2.10.0-py2.py3-none-any.whl", hash = "sha256:8a4eaf0d0495c7395bdab3589ac2db602797d76207242c17d470186815706610"},
+    {file = "pycodestyle-2.10.0.tar.gz", hash = "sha256:347187bdb476329d98f695c213d7295a846d1152ff4fe9bacb8a9590b8ee7053"},
+]
+
+[[package]]
+name = "pyflakes"
+version = "3.0.1"
+description = "passive checker of Python programs"
+optional = false
+python-versions = ">=3.6"
+files = [
+    {file = "pyflakes-3.0.1-py2.py3-none-any.whl", hash = "sha256:ec55bf7fe21fff7f1ad2f7da62363d749e2a470500eab1b555334b67aa1ef8cf"},
+    {file = "pyflakes-3.0.1.tar.gz", hash = "sha256:ec8b276a6b60bd80defed25add7e439881c19e64850afd9b346283d4165fd0fd"},
+]
+
+[metadata]
+lock-version = "2.0"
+python-versions = "^3.8.1"
+content-hash = "ae9182c39898ce95ac2f3cc63f487560d954e9af3220b5e508cf2ea23442faa1"

pyproject.toml

@@ -0,0 +1,24 @@
+[tool.poetry]
+name = "wg-bootstrap"
+version = "0.1.0"
+description = "wg-quick wrapper for creating interface config and up/down wg"
+authors = ["MOIS3Y <s.zhukovskii@ispsystem.com>"]
+license = "GPL-3.0-or-later"
+repository = "https://git.isptech.ru/ISPsystem/wg-bootstrap"
+keywords = ["wireguard-tools", "wg", "interface"]
+include = ["CHANGELOG.md"]
+readme = "README.md"
+packages = [{include = "wg_bootstrap"}]
+
+[tool.poetry.dependencies]
+python = "^3.8.1"
+
+[tool.poetry.group.dev.dependencies]
+flake8 = "^6.0.0"
+
+[tool.poetry.scripts]
+wg-start = "wg_bootstrap.wg_start:main"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

wg_bootstrap/wg_start.py

@@ -0,0 +1,199 @@
+#!/usr/bin/env python3
+import os
+import sys
+import argparse
+import subprocess as sp
+from pathlib import Path
+from functools import wraps
+
+__author__ = "MOIS3Y"
+__version__ = "0.1.0"
+
+
+# check development or production environment:
+if os.environ.get("WG_BOOTSTRAP"):
+    CONFIG_PATH = Path(__file__).resolve().parents[1] / "etc" / "wireguard"
+else:
+    CONFIG_PATH = Path("/etc/wireguard")
+
+
+def sudo_required(func):
+    """
+    the decorator raise an exception PermissionError
+    when trying to call a function without superuser privileges (sudo)
+    or not from under the root.
+
+    Args:
+        func (_function_): any function
+    """
+    @wraps(func)  # <- save meta info for wrapped func
+    def is_root(*args, **kwargs):
+        if not os.environ.get("SUDO_UID") and os.geteuid() != 0:
+            sys.tracebacklimit = 0
+            prog = os.path.basename(__file__)
+            error_message = f"You need to run {prog} with sudo or as root."
+            raise PermissionError(error_message)
+        result = func(*args, **kwargs)
+        return result
+
+    return is_root
+
+
+@sudo_required
+def get_wg_conf() -> dict:
+    """
+    gets the values for the settings dict interactively from the user input
+
+    Returns:
+        dict: settings for wg interface
+    """
+    settings: dict = {
+        "Interface": {
+            "PrivateKey": None,
+            "Address": None,
+            "DNS": None
+        },
+        "Peer": {
+            "PublicKey": None,
+            "AllowedIPs": None,
+            "Endpoint": None
+        }
+    }
+    for title in settings:
+        for key, value in settings[title].items():
+            settings[title][key] = input(f"[{title}] - {key}: ")
+
+    return settings
+
+
+@sudo_required
+def gen_wg_conf(config_path, **settings):
+    """
+    generates a config file for the wireguard interface, 
+    receiving a dict (**settings) with the desired values
+
+    Args:
+        config_path (Path): path to wg interface configuration file
+        **settings (dict): settings for wg interface
+    """
+    # check if interface parent dir exist:
+    config_path.parent.mkdir(parents=True, exist_ok=True)
+
+    with open(config_path, 'w') as file:
+        for title in settings:
+            if title == "Peer":
+                file.write(f"\n[{title}]\n")
+            else:
+                file.write(f"[{title}]\n")
+
+            for key, value in settings[title].items():
+                file.write(f"{key} = {value}\n")
+
+
+@sudo_required
+def use_wg_conf(config_path, command):
+    sp.run([command, config_path])
+
+
+def wg_quick(interface, command):
+    sp.run(["wg-quick", command, interface])
+
+
+def exist_wg_conf(config_path):
+    return config_path.exists()
+
+
+@sudo_required
+def change_allowed_ips(config_path, allowed_ips, action="replace"):
+    """
+    the function changes the AllowedIPs value in 
+    the passed configuration file.
+
+    Args:
+        config_file (Path): path to wg interface configuration file
+        allowed_ips (str): comma and space separated ip addresses
+        action (str, optional): replace or add. Defaults to "replace".
+    """
+    with open(config_path, 'r') as source:
+        lines = source.readlines()
+
+    with open(config_path, 'w') as target:
+        for line in lines:
+            if line.strip().startswith('AllowedIPs = '):
+                if action == "add":
+                    old = line.strip('\n')
+                    line = f"{old}, {allowed_ips}\n"
+                else:
+                    line = f"AllowedIPs = {allowed_ips}\n"
+            target.write(line)
+
+
+def create_parser():
+    parser = argparse.ArgumentParser(
+        description="CRUD WireGuard config file or UP/DOWN wg-interface"
+    )
+    parser.add_argument(
+        "command",
+        type=str,
+        choices=["init", "cat", "rm", "add", "replace", "up", "down"],
+        help="Action with interface"
+    )
+    parser.add_argument(
+        "interface",
+        type=str,
+        help="WG interface name (wg0, wg1, wgName etc...)"
+    )
+    parser.add_argument(
+        "allowedIPs",
+        type=str,
+        nargs="?",
+        help="Set AllowedIPs must be a string"
+    )
+    parser.add_argument(
+        "-v",
+        "--version",
+        action='version',
+        version=f"%(prog)s - {__version__}")
+
+    return parser
+
+
+def main():
+    # init cmd args parser:
+    parser = create_parser()
+    args = parser.parse_args()
+
+    # current wg interface (get from required arg from user):
+    config_path = CONFIG_PATH / f"{args.interface}.conf"
+
+    # interface config:
+    if args.command == "init":
+        settings = get_wg_conf()
+        gen_wg_conf(config_path, **settings)
+
+    if args.command == "cat" or args.command == "rm":
+        if not exist_wg_conf(config_path):
+            raise parser.error(f"{config_path} does not exist")
+        use_wg_conf(config_path, args.command)
+
+    if args.command == "add" or args.command == "replace":
+        if not args.allowedIPs:
+            raise parser.error("With add/replace you must pass AllowedIPs")
+        if not exist_wg_conf(config_path):
+            raise parser.error(f"{config_path} does not exist")
+        change_allowed_ips(
+            config_path,
+            args.allowedIPs,
+            args.command
+        )
+
+    # run wg-quick:
+    if args.command == "up" or args.command == "down":
+        try:
+            wg_quick(args.interface, args.command)
+        except KeyboardInterrupt or Exception:
+            sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()