initial commit

.gitea/workflows/build.yaml

@@ -0,0 +1,27 @@
+name: CI
+on:
+  push:
+    branches: [ master ]
+jobs:
+  build:
+    runs-on: woryzen
+    steps:
+      - name: Checkout sources
+        uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          cache: 'pip'
+      - name: Create virtualenv
+        run: |
+          python -m venv .venv
+          .venv/bin/pip install -r requirements-dev.txt
+      - name: Execute build
+        run: |
+          .venv/bin/python -m build
+      - name: Publish artifacts
+        env:
+          TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
+          TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
+          TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
+        run: |
+          .venv/bin/python -m twine upload --repository gitea dist/*{.whl,tar.gz}

.gitignore

@@ -0,0 +1,5 @@
+.venv
+.idea
+*.pyc
+*.egg-info
+__pycache__

conf/example.toml

@@ -0,0 +1,17 @@
+interface_name = "docker-cluster"
+netmask = "10.45.62.0/24"
+dns = [ "192.168.78.7", "1.1.1.1"]
+
+[[server]]
+name = "odroid-c4"
+address = '155.24.76.11'
+port = 1234
+
+[[peer]]
+name = "odroid-hc1"
+
+[[peer]]
+name = "woryzen"
+
+[[peer]]
+name = "sugo14"

pyproject.toml

@@ -0,0 +1,53 @@
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "wg_builder"
+version = "0.0.1"
+authors = [
+  { name="Walter Oggioni", email="oggioni.walter@gmail.com" },
+]
+description = "Wireguard network builder"
+readme = "README.md"
+requires-python = ">=3.12"
+classifiers = [
+        'Development Status :: 3 - Alpha',
+        'Topic :: Utilities',
+        'License :: OSI Approved :: MIT License',
+        'Intended Audience :: System Administrators',
+        'Intended Audience :: Developers',
+        'Environment :: Console',
+        'License :: OSI Approved :: MIT License',
+        'Programming Language :: Python :: 3',
+]
+dependencies = [
+    'python-wireguard',
+    'toml',
+    'netaddr',
+    'typing_extensions==4.7.1',
+    'pwo'
+]
+
+[project.optional-dependencies]
+dev = [
+    "build", "pip-tools", "mypy", "ipdb"
+]
+
+
+[project.urls]
+"Homepage" = "https://gitea.woggioni.net/woggioni/wg_builder"
+"Bug Tracker" = "https://gitea.woggioni.net/woggioni/wg_builder/issues"
+
+[project.scripts]
+wg-builder = "wg_builder.cli:main"
+
+[tool.mypy]
+python_version = "3.12"
+disallow_untyped_defs = true
+show_error_codes = true
+no_implicit_optional = true
+warn_return_any = true
+warn_unused_ignores = true
+exclude = ["scripts", "docs", "test"]
+strict = true

requirements-dev.txt

@@ -0,0 +1,84 @@
+#
+# This file is autogenerated by pip-compile with Python 3.12
+# by the following command:
+#
+#    pip-compile --extra=dev --output-file=requirements-dev.txt --strip-extras pyproject.toml
+#
+--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple/
+--extra-index-url https://pypi.org/simple
+
+asttokens==2.4.1
+    # via stack-data
+build==1.2.2
+    # via
+    #   pip-tools
+    #   wg_builder (pyproject.toml)
+click==8.1.7
+    # via pip-tools
+decorator==5.1.1
+    # via
+    #   ipdb
+    #   ipython
+executing==2.1.0
+    # via stack-data
+ipdb==0.13.13
+    # via wg_builder (pyproject.toml)
+ipython==8.27.0
+    # via ipdb
+jedi==0.19.1
+    # via ipython
+matplotlib-inline==0.1.7
+    # via ipython
+mypy==1.11.2
+    # via wg_builder (pyproject.toml)
+mypy-extensions==1.0.0
+    # via mypy
+netaddr==1.3.0
+    # via wg_builder (pyproject.toml)
+packaging==24.1
+    # via build
+parso==0.8.4
+    # via jedi
+pexpect==4.9.0
+    # via ipython
+pip-tools==7.4.1
+    # via wg_builder (pyproject.toml)
+prompt-toolkit==3.0.47
+    # via ipython
+ptyprocess==0.7.0
+    # via pexpect
+pure-eval==0.2.3
+    # via stack-data
+pwo==0.0.2
+    # via wg_builder (pyproject.toml)
+pygments==2.18.0
+    # via ipython
+pyproject-hooks==1.1.0
+    # via
+    #   build
+    #   pip-tools
+python-wireguard==0.2.2
+    # via wg_builder (pyproject.toml)
+six==1.16.0
+    # via asttokens
+stack-data==0.6.3
+    # via ipython
+toml==0.10.2
+    # via wg_builder (pyproject.toml)
+traitlets==5.14.3
+    # via
+    #   ipython
+    #   matplotlib-inline
+typing-extensions==4.7.1
+    # via
+    #   mypy
+    #   pwo
+    #   wg_builder (pyproject.toml)
+wcwidth==0.2.13
+    # via prompt-toolkit
+wheel==0.44.0
+    # via pip-tools
+
+# The following packages are considered to be unsafe in a requirements file:
+# pip
+# setuptools

requirements.txt

@@ -0,0 +1,21 @@
+#
+# This file is autogenerated by pip-compile with Python 3.12
+# by the following command:
+#
+#    pip-compile --output-file=requirements.txt --strip-extras pyproject.toml
+#
+--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple/
+--extra-index-url https://pypi.org/simple
+
+netaddr==1.3.0
+    # via wg_builder (pyproject.toml)
+pwo==0.0.2
+    # via wg_builder (pyproject.toml)
+python-wireguard==0.2.2
+    # via wg_builder (pyproject.toml)
+toml==0.10.2
+    # via wg_builder (pyproject.toml)
+typing-extensions==4.7.1
+    # via
+    #   pwo
+    #   wg_builder (pyproject.toml)

src/wg_builder/builder.py

@@ -0,0 +1,143 @@
+import tarfile
+from base64 import b64encode
+from os import makedirs
+from pathlib import Path
+from random import SystemRandom
+from tarfile import TarInfo
+from textwrap import dedent
+
+import toml
+from netaddr import IPNetwork
+from netaddr.ip import IPAddress
+from pwo import Maybe
+from python_wireguard import Key
+
+random = SystemRandom()
+from io import BytesIO
+import codecs
+
+def genPsk() -> str:
+    return b64encode(random.randbytes(32)).decode('UTF-8')
+
+
+def build_network(input_file: Path = Path('/dev/stdin'), output_folder: Path = Path('.'), archive=False, **kwargs):
+    with open(input_file) as conf:
+        configuration = toml.load(conf)
+
+    if 'dns' in configuration:
+        dns = tuple(IPAddress(address) for address in configuration['dns'])
+    else:
+        dns = tuple()
+
+    network_name = Maybe.of_nullable(configuration.get('network_name')).or_else(input_file.stem)
+    network = IPNetwork(configuration['netmask'])
+    address_it = network.iter_hosts()
+
+    class Server:
+        endpoint: str
+        address: str
+        port: int
+        private_key: str
+        public_key: str
+
+        def __init__(self, conf):
+            self.name = conf['name']
+            self.private_key, self.public_key = Key.key_pair()
+            self.address = str(address_it.__next__())
+            self.endpoint = conf['address']
+            self.port = conf['port']
+
+    class Peer:
+        name: str
+        private_key: str
+        public_key: str
+        allowed_ip: str
+
+        def __init__(self, conf):
+            self.name = conf['name']
+            self.private_key, self.public_key = Key.key_pair()
+            self.allowed_ip = str(address_it.__next__())
+
+    servers = [Server(server) for server in configuration['server']]
+
+    peers = [Peer(peer) for peer in configuration['peer']]
+
+    pre_shared_keys = {(server, peer): genPsk() for peer in peers for server in servers}
+
+
+    def write_server_conf(server, file_obj):
+        text = dedent(f"""\
+            [Interface]
+            Address = {server.address}
+            ListenPort = {server.port}
+            PrivateKey = {str(server.private_key)}
+        """)
+        file_obj.write(text)
+        for peer in peers:
+            text = dedent(f"""\
+
+                [Peer]
+                PublicKey = {peer.public_key}
+                AllowedIPs = {peer.allowed_ip} 
+                PresharedKey = {pre_shared_keys[(server, peer)]}           
+            """)
+            file_obj.write(text)
+
+    def write_client_conf(peer, file_obj):
+        text = dedent(f"""\
+            [Interface]
+            Address = {peer.allowed_ip}/32
+            PrivateKey = {str(peer.private_key)}
+        """)
+        dns_config = (Maybe.of(dns)
+                      .filter(lambda it: len(it) > 0)
+                      .map(lambda it: f'DNS = {', '.join((str(addr) for addr in it))}\n')
+                      .or_else(''))
+        file_obj.write(text + dns_config)
+        for server in servers:
+            text = dedent(f"""\
+
+                [Peer]
+                PublicKey = {server.public_key}
+                PresharedKey = {pre_shared_keys[(server, peer)]}
+                Endpoint = {server.endpoint}:{server.port}
+                AllowedIPs = {str(network)}            
+            """)
+            file_obj.write(text)
+
+    makedirs(output_folder, exist_ok=True)
+    for server in servers:
+        if archive:
+            with tarfile.open(output_folder / f'{server.name}.tar.gz', 'w:gz') as archive:
+                with BytesIO() as bio:
+                    writer = codecs.getwriter('utf-8')(bio)
+                    write_server_conf(server, writer)
+                    name = f'/etc/wireguard/{network_name}.conf'
+                    tarinfo = TarInfo(name)
+                    tarinfo.size = bio.tell()
+                    bio.seek(0)
+                    archive.addfile(tarinfo, bio)
+        else:
+            server_folder = output_folder / f'{server.name}'
+            makedirs(server_folder, exist_ok=True)
+            with (open(server_folder / f'{network_name}.conf', 'w') as f):
+                write_server_conf(server, f)
+
+    for peer in peers:
+        if archive:
+            with tarfile.open(output_folder / f'{peer.name}.tar.gz', 'w:gz') as archive:
+                with BytesIO() as bio:
+                    writer = codecs.getwriter('utf-8')(bio)
+                    write_client_conf(peer, writer)
+                    name = f'/etc/wireguard/{network_name}.conf'
+                    tarinfo = TarInfo(name)
+                    tarinfo.size = bio.tell()
+                    bio.seek(0)
+                    archive.addfile(tarinfo, bio)
+        else:
+            peer_folder = output_folder / f'{peer.name}'
+            makedirs(peer_folder, exist_ok=True)
+            with (open(peer_folder / f'{network_name}.conf', 'w') as f):
+                write_client_conf(peer, f)
+
+

src/wg_builder/cli.py

@@ -0,0 +1,35 @@
+from typing_extensions import Optional, List
+
+import argparse
+import sys
+
+from pathlib import Path
+from .builder import build_network
+
+def main(args: Optional[List[str]] = None):
+    parser = argparse.ArgumentParser(description="A simple CLI program to create Wireguard networks")
+    parser.add_argument(
+        '-i',
+        '--input-file',
+        help='Input file (defaults to stdin)',
+        type=Path,
+    )
+    parser.add_argument(
+        '-o',
+        '--output-folder',
+        help='Output folder (defaults to .)',
+        type=Path,
+    )
+    parser.add_argument(
+        '-a',
+        '--archive',
+        help='store the configuration in archive files',
+        action='store_true',
+    )
+    parser.set_defaults(func=build_network)
+    args = parser.parse_args(args)
+    args.func(**vars(args))
+
+
+if __name__ == '__main__':
+    main(sys.argv)