formatting, minor tweaks, fixed txpo

This commit is contained in:
Kristian Krsnik 2023-08-03 23:14:04 +02:00
parent 67e4afd0bf
commit e4f94e04a6
5 changed files with 55 additions and 47 deletions

2
.gitignore vendored
View File

@ -2,3 +2,5 @@ result
config.json config.json
log.txt log.txt
__pycache__/ __pycache__/
.vscode/
.direnv/

View File

@ -1,6 +1,10 @@
import requests, json, re, os import requests
import json
import re
import os
from datetime import datetime from datetime import datetime
def loadSettings(path: str = f"{os.path.expanduser('~')}/.config/dyn-gandi/config.json") -> dict: def loadSettings(path: str = f"{os.path.expanduser('~')}/.config/dyn-gandi/config.json") -> dict:
# TODO: check integrity # TODO: check integrity
try: try:
@ -15,32 +19,33 @@ def loadSettings(path: str = f"{os.path.expanduser('~')}/.config/dyn-gandi/confi
with open(key, 'r') as file: with open(key, 'r') as file:
api_key = file.readline().strip() api_key = file.readline().strip()
mask[key] = api_key mask[key] = api_key
config['api'] = { mask[key]: value for key, value in config['api'].items() } config['api'] = {mask[key]: value for key,
value in config['api'].items()}
except FileNotFoundError: except FileNotFoundError:
sample_config = { quit()
'api': {}, # TODO log error and remove code below (do not create an unwanted config)
'ttl': 3600,
'resolvers': [ 'https://ifconfig.me' ],
'log_path': './log.txt'
}
with open('config.json', 'w') as file:
json.dump(sample_config, file, indent = 2)
return sample_config def anewfunction():
for i, key in range(12):
print(i)
# Appends logline at the end of the file file with a timestamp
def log(logline: str, path: str = './log.txt') -> None: def log(logline: str, path: str = './log.txt') -> None:
# Appends logline at the end of the file file with a timestamp
with open(path, 'a') as file: with open(path, 'a') as file:
file.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}]{logline}\n") file.write(
f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}]{logline}\n")
# Go through resolvers until one retuns an IP
def getCurrentIP(resolvers: list[str], log: callable) -> None | str: def getCurrentIP(resolvers: list[str], log: callable) -> None | str:
# Go through resolvers until one retuns an IP
for resolver in resolvers: for resolver in resolvers:
response = requests.get(resolver) response = requests.get(resolver)
if response.status_code == 200: if response.status_code == 200:
current_ip = response.text.strip() current_ip = response.text.strip()
is_ipv4 = re.search('^((25[0-5]|(2[0-4]|1\d|[1-9]|)\d)\.?\\b){4}$', current_ip) is not None is_ipv4 = re.search(
'^((25[0-5]|(2[0-4]|1\d|[1-9]|)\d)\.?\\b){4}$', current_ip) is not None
if is_ipv4: if is_ipv4:
log(f"[OK][{resolver}] Current IP: '{current_ip}'") log(f"[OK][{resolver}] Current IP: '{current_ip}'")
return current_ip # Break if we have found our IP return current_ip # Break if we have found our IP
@ -50,27 +55,31 @@ def getCurrentIP(resolvers: list[str], log: callable) -> None | str:
log(f"[ERROR][{resolver}][{response.status_code}] {response.text}") log(f"[ERROR][{resolver}][{response.status_code}] {response.text}")
return None return None
# Returns the records that need to be renewed
# TODO: Also detect TTL change
def checkRecords(api: dict, current_ip: str, log: callable) -> dict: def checkRecords(api: dict, current_ip: str, log: callable) -> dict:
# Returns the records that need to be renewed
# TODO: Also detect TTL change
records_to_renew = api records_to_renew = api
for key, domains in api.items(): for key, domains in api.items():
headers = { 'authorization': f"Apikey {key}" } headers = {'authorization': f"Apikey {key}"}
for domain, records in domains.items(): for domain, records in domains.items():
response = requests.get(f"https://api.gandi.net/v5/livedns/domains/{domain}/records?rrset_type=A", headers = headers) response = requests.get(
f"https://api.gandi.net/v5/livedns/domains/{domain}/records?rrset_type=A", headers=headers)
if response.status_code == 200: if response.status_code == 200:
records_to_renew[key][domain] = list(set(records_to_renew[key][domain]) - set(record['rrset_name'] for record in (filter(lambda x: current_ip in x['rrset_values'], json.loads(response.json()))))) records_to_renew[key][domain] = list(set(records_to_renew[key][domain]) - set(record['rrset_name'] for record in (
filter(lambda x: current_ip in x['rrset_values'], json.loads(response.json())))))
else: else:
log(f"[ERROR][{domain}][{response.status_code}] {response.json()}") log(f"[ERROR][{domain}][{response.status_code}] {response.json()}")
return records_to_renew return records_to_renew
# Updates the records and reates them if they don't exist
def renewRecords(records, current_ip, log, ttl = 3600): def renewRecords(records, current_ip, log, ttl=3600):
# Updates the records and reates them if they don't exist
payload = json.dumps({ payload = json.dumps({
'items': [{ 'items': [{
'rrset_type': 'A', 'rrset_type': 'A',
'rrset_values': [ current_ip ], 'rrset_values': [current_ip],
'rrset_ttl': ttl 'rrset_ttl': ttl
}] }]
}) })
@ -83,21 +92,24 @@ def renewRecords(records, current_ip, log, ttl = 3600):
for domain, records in domains.items(): for domain, records in domains.items():
for record in records: for record in records:
response = requests.put(f"https://api.gandi.net/v5/livedns/domains/{domain}/records/{record}", data = payload, headers = headers) response = requests.put(
f"https://api.gandi.net/v5/livedns/domains/{domain}/records/{record}", data=payload, headers=headers)
if response.status_code == 201: if response.status_code == 201:
log(f"[OK][{domain}][{record}] Renewed with IP '{current_ip}'") log(f"[OK][{domain}][{record}] Renewed with IP '{current_ip}'")
else: else:
log(f"[ERROR][{domain}][{record}][{response.status_code}] {response.json()}") log(f"[ERROR][{domain}][{record}][{response.status_code}] {response.json()}")
def main(): def main():
settings = loadSettings() settings = loadSettings()
log_path = settings['log_path'] log_path = settings['log_path']
ttl = settings['ttl'] ttl = settings['ttl']
myLog = lambda x: log(x, log_path) def myLog(x): return log(x, log_path)
current_ip = getCurrentIP(settings['resolvers'], myLog) current_ip = getCurrentIP(settings['resolvers'], myLog)
records_to_renew = checkRecords(settings['api'], current_ip, myLog) records_to_renew = checkRecords(settings['api'], current_ip, myLog)
renewRecords(records_to_renew, current_ip, myLog, ttl = ttl) renewRecords(records_to_renew, current_ip, myLog, ttl=ttl)
if __name__ == '__main__': if __name__ == '__main__':
main() main()

View File

@ -2,13 +2,13 @@
description = "A very basic flake"; description = "A very basic flake";
inputs = { inputs = {
nixpkgs.url = github:nixos/nixpkgs/nixos-23.05; nixpkgs.url = "github:nixos/nixpkgs/nixos-23.05";
}; };
outputs = { outputs = {
self, self,
nixpkgs, nixpkgs,
} @ inputs: let }: let
supportedSystems = ["x86_64-linux" "x86_64-darwin" "aarch64-linux" "aarch64-darwin"]; supportedSystems = ["x86_64-linux" "x86_64-darwin" "aarch64-linux" "aarch64-darwin"];
forAllSystems = nixpkgs.lib.genAttrs supportedSystems; forAllSystems = nixpkgs.lib.genAttrs supportedSystems;
pkgs = forAllSystems (system: nixpkgs.legacyPackages.${system}); pkgs = forAllSystems (system: nixpkgs.legacyPackages.${system});
@ -24,8 +24,6 @@
# Dev Shell # Dev Shell
devShells = forAllSystems (system: { devShells = forAllSystems (system: {
default = pkgs.${system}.mkShellNoCC { default = pkgs.${system}.mkShellNoCC {
shellHook = "echo Welcome to your nix-powered dev environment for dyn-gandi!";
packages = with pkgs.${system}; [ packages = with pkgs.${system}; [
(poetry2nix.mkPoetryEnv {projectDir = self;}) (poetry2nix.mkPoetryEnv {projectDir = self;})
poetry poetry

View File

@ -33,13 +33,7 @@ in {
in in
valueType; valueType;
default = {api = {"example.com" = ["@" "www"];};}; default = {};
example = {
General = {
disabledTrayIcon = true;
showStartupLaunchMessage = false;
};
};
description = ''''; description = '''';
}; };
}; };

View File

@ -1,16 +1,18 @@
inputs: { config, lib, pkgs, ... }: inputs: {
let config,
lib,
pkgs,
...
}: let
cfg = config.services.dyn-gandi; cfg = config.services.dyn-gandi;
inherit (lib) mkIf mkEnableOption; inherit (lib) mkIf mkEnableOption;
#package = inputs.self.packages.${pkgs.stdenv.hostPlatform.system}.default; #package = inputs.self.packages.${pkgs.stdenv.hostPlatform.system}.default;
in { in {
options.services.dyn-gandi = { options.services.dyn-gandi = {
enable = mkEnableOption null; enable = mkEnableOption null;
}; };
config = mkIf cfg.enable { config = mkIf cfg.enable {
environment.systemPackages = [ cfg.package ]; environment.systemPackages = [cfg.package];
} };
} }