(wanna be) full dns stack ansible role
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

147 lines
4.3 KiB

#!/usr/bin/env python
import argparse
import os
import subprocess
import sys
import typing as T
import yaml
ENCODING = "utf-8"
HOSTS: T.Dict[T.Text, T.Dict[T.Text, T.Text]] = {}
# From: https://stackoverflow.com/a/43060743
class DummyVault(yaml.YAMLObject):
yaml_tag = "!vault"
def __init__(self, cyphertext):
self.cyphertext = "\n".join([e.strip() for e in cyphertext.split("\n")])
def __repr__(self):
return f"{self.__class__.__name__}({self.cyphertext[:10]}...)"
@classmethod
def from_yaml(cls, loader, node):
return DummyVault(node.value)
@classmethod
def to_yaml(cls, dumper, data):
return dumper.represent_scalar(cls.yaml_tag, data.cyphertext, style="|")
yaml.SafeLoader.add_constructor("!vault", DummyVault.from_yaml)
yaml.SafeDumper.add_multi_representer(DummyVault, DummyVault.to_yaml)
def load_hosts(inventory: T.Text, role: T.Optional[T.Text]) -> T.List[T.Text]:
with open(inventory, "r") as f:
data = yaml.load(f, Loader=yaml.SafeLoader)
if role:
return [k for k in data["all"]["children"][role]["hosts"].keys()]
return [k for k in data["all"]["hosts"].keys()]
def gen_key(name: T.Text) -> None:
with open(f"{name}.sec", "w") as sec:
subprocess.call(["wg", "genkey"], stdout=sec)
with open(f"{name}.sec", "rb", 0) as sec_r, open(f"{name}.pub", "w") as pub:
subprocess.call(["wg", "pubkey"], stdin=sec_r, stdout=pub)
def _encrypt_string(
plaintext: T.Text,
passfile: T.Text,
vault_id: T.Optional[T.Text],
passfile_dir: T.Optional[T.Text],
) -> bytes:
if passfile_dir is None:
passfile_dir = os.getcwd()
if vault_id is None:
return _encrypt_string_simple(plaintext, passfile, passfile_dir)
else:
return _encrypt_string_vault(plaintext, f"{vault_id}@{passfile}", passfile_dir)
def _encrypt_string_simple(
plaintext: T.Text, passfile: T.Text, passfile_dir: T.Text
) -> bytes:
return subprocess.check_output(
[
"ansible-vault",
"encrypt_string",
f"--vault-password-file={passfile}",
plaintext,
],
cwd=passfile_dir,
)
def _encrypt_string_vault(
plaintext: T.Text, vault_passfile: T.Text, passfile_dir: T.Text
) -> bytes:
return subprocess.check_output(
["ansible-vault", "encrypt_string", f"--vault-id={vault_passfile}", plaintext],
cwd=passfile_dir,
)
def to_vault(
name: T.Text,
passfile: T.Text,
vault_id: T.Optional[T.Text],
passfile_dir: T.Optional[T.Text],
) -> None:
with open(f"{name}.pub", "r") as pub:
pubkey = pub.readline().strip("\n")
enc_pub = _encrypt_string(pubkey, passfile, vault_id, passfile_dir)
with open(f"{name}.sec", "r") as sec:
seckey = sec.readline().strip("\n")
enc_sec = _encrypt_string(seckey, passfile, vault_id, passfile_dir)
HOSTS[name] = {
"public_key": yaml.load(enc_pub.decode(ENCODING), Loader=yaml.SafeLoader),
"private_key": yaml.load(enc_sec.decode(ENCODING), Loader=yaml.SafeLoader),
}
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Generates keys for wireguard connection of the dns servers"
)
parser.add_argument(
"inventory", metavar="INVENTORY", type=str, help="path to the inventory"
)
parser.add_argument(
"passfile",
metavar="VAULT_PASSFILE",
type=str,
help="the name of the file that contains the passphrase for the inventory",
)
parser.add_argument(
"--vault-id",
metavar="VAULT_ID",
type=str,
help="the name of the (existing) vault",
)
parser.add_argument(
"--passfile-dir",
metavar="PASSFILE_DIR",
type=str,
help="path where the passfile is located",
)
parser.add_argument(
"--role",
metavar="ROLE",
type=str,
help="an optional key to use to filter the search for hosts in the inventory",
)
args = parser.parse_args()
for host in load_hosts(args.inventory, args.role):
gen_key(host)
to_vault(host, args.passfile, args.vault_id, args.passfile_dir)
result = yaml.dump(HOSTS, Dumper=yaml.SafeDumper)
with open("result.yml", "w") as res:
res.write(result)
print(result)