repo-cloner/repo_cloner/initialization_wizard.py

354 lines
10 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
import logging
from repo_cloner.lib.logger_setup import log
import os
import pyinputplus as pyip
from typing import Optional, Callable
from pathlib import Path
import subprocess
from repo_cloner.process_repository_dir import clone_or_fetch
# base dir
base_dir: Optional[Path] = None
cloner_prefix: str = "cloner-"
data: dict = {}
initialized_folder = None
# determine starting user and devote UID/GID to unprivileged user - safety first :)
def check_privileges():
if os.getuid() == 0:
log.info(f"Running as root, downgrading permissions")
os.setgid(1000)
os.setuid(1000)
os.setegid(1000)
os.seteuid(1000)
log.info(f"New UID:GID: {os.getuid()}:{os.getgid()}")
def parse_args():
global base_dir
import argparse
# parse input arguments
parser = argparse.ArgumentParser(description = "repo-cloner initialization wizzard")
parser.add_argument(
'--base-dir', help = 'path to directory containing whole cloner structure', required = True,
default = None, type = str)
parser.add_argument('--colored', help = "enable colored log output even tty() is not detected")
args = parser.parse_args()
base_dir = Path(args.base_dir)
def gen_config_file(
conf_dir: Path,
cloner_repo_url,
cloner_project_name,
cloner_interval,
cloner_submodules,
cloner_submodule_depth,
detector,
**kwargs
):
from datetime import datetime
log.info(f"Creating config files in {conf_dir.as_posix()}")
if not conf_dir.exists():
log.info(f"Creating config dir")
conf_dir.mkdir(parents = True)
log.info("Creating cloner.cfg")
conf_file = conf_dir.joinpath("cloner.cfg")
conf_file.write_text(
f"# cloner.cfg\n"
"# main config\n"
f"# created at {datetime.now().strftime('%Y-%m-%d %X')}\n\n"
"# main url - url of main repo - just to clone\n"
f"cloner_repo_url = {cloner_repo_url}\n\n"
"# project name (names of volumes are derrived from this\n"
f"cloner_project_name = {cloner_project_name}\n\n"
"# cloner interval (in minutes, default=0 - run always)\n"
f"cloner_interval = {cloner_interval}\n\n"
"# do you need submodules support? (1/0)\n"
f"cloner_submodules = {1 if cloner_submodules else 0}\n\n"
"# max depth of submodule scan (default = unlimited, uncomment to use)\n"
f"{'' if cloner_submodules else '# '}cloner_submodule_depth = {cloner_submodule_depth}\n\n"
)
if detector:
log.info("Creating detector.cfg")
conf_file = conf_dir.joinpath("detector.cfg")
conf_file.write_text(
"# this file is config for detector\n"
"# now, it is empty - to disable detector, just delete it!\n"
)
def check_project_name(name: str):
if not len(name):
raise Exception("Empty input is invalid input!")
target = os.path.join(base_dir, f"{cloner_prefix}{name}")
log.debug(f"Validating project name - path {target}")
if os.path.exists(target):
log.warning(f"Project name occupied: {target}")
raise Exception(f"Project name {name} is occupied by another project")
def check_url(name: str):
if not len(name):
raise Exception("You must input URL")
def input_default_str(query: str, default_value: str, validation: Optional[Callable[[str], None]] = None) -> str:
log.debug(f"Input query {query} with default {default_value}")
while True:
new_query = query
if len(default_value):
new_query += f" [{default_value}] "
else:
new_query += " "
ret = pyip.inputStr(new_query, blank = True, strip = True)
if ret == "":
log.debug(f"Empty query answer => using previous/default value")
ret = default_value
try:
if validation:
validation(ret)
except Exception as e:
log.warning(e.__str__())
default_value = ret
continue
log.debug("Query finished")
break
return ret
def input_default_int(query: str, default_value: int, validation: Optional[Callable[[int], None]] = None) -> int:
log.debug(f"Input query {query} with default {default_value}")
ret = default_value
while True:
new_query = f"{query} [{default_value}] "
ret = pyip.inputInt(new_query, blank = True, strip = True, min = 0)
if not ret and not ret == 0:
log.debug(f"Empty query answer => using previous/default value")
ret = default_value
try:
if validation:
validation(ret)
except Exception as e:
log.warning(e.__str__())
default_value = ret
continue
log.debug("Query finished")
break
return ret
def input_default_bool(query: str, default_value: bool) -> bool:
log.debug(f"Input query {query} with default {default_value}")
new_query = f"{query} [{'Y' if default_value else 'N'}] "
ret = pyip.inputYesNo(new_query, blank = True, strip = True)
if not len(ret):
log.debug(f"Empty query answer => using previous/default value")
ret = default_value
if ret == "yes":
ret = True
if ret == "no":
ret = False
log.debug("Query finished")
return ret
def query_repo_info() -> bool:
log.debug(f"Querying base info")
if not initialized_folder:
# project name
data["cloner_project_name"] = input_default_str(
"Enter project name:",
data["cloner_project_name"],
check_project_name)
else:
log.warning(f"Project folder was initialized -> unable to change")
# url
data["cloner_repo_url"] = input_default_str("Enter project url:", data["cloner_repo_url"], check_url)
# interval
data["cloner_interval"] = input_default_int("Enter sync interval:", data["cloner_interval"])
# submodule support
data["cloner_submodules"] = input_default_bool("Mirror including submodules? [y/n]:", data["cloner_submodules"])
if data["cloner_submodules"]:
data["cloner_submodule_depth"] = input_default_int(
"Limit for submodule discovery [<number>/0]:",
data["cloner_submodule_depth"]
)
data["detector"] = input_default_bool("Do you want to enable CI support? (detector) [y/n]", data["detector"])
data["detector_init"] = input_default_bool("Do you want to init CI detector caches? [y/n]", data["detector_init"])
def query_repo_info_recursive():
while True:
query_repo_info()
print("Actual settings:")
for key, value in data.items():
print(f"{key: <30} : {value}")
if not input_default_bool("Do you want to edit config? [y/n]", False):
break
def create_ssh_key(ssh_dir: Path):
log.info(f"Creating SSH key...")
keyfile = ssh_dir.joinpath("identity")
subprocess.run([
"/usr/bin/ssh-keygen",
"-f", keyfile.as_posix(),
"-t", "ed25519",
"-C", f"cloner-deploy-key-{data['cloner_project_name']}"
, "-N", ""
])
public = ssh_dir.joinpath("identity.pub").read_text().strip()
print("Public key is:")
print("-----------------------------------------------------")
print(public)
print("-----------------------------------------------------")
print("Please make sure that key is set up at your git hosting and press enter..")
pyip.inputStr("", blank = True)
def reuse_ssh_key(ssh_dir: Path):
log.info("Reusing any ssh key")
key_contents = ""
print("Paste password-less private key here. End with empty line:")
while True:
ret = pyip.inputStr("", blank = True)
if not len(ret):
break
key_contents += f"{ret}\n"
# public keyfile
keyfile_pub = ssh_dir.joinpath("identity.pub")
if keyfile_pub.exists():
log.debug(f"Removing old public key")
keyfile_pub.unlink()
# private keyfile
keyfile = ssh_dir.joinpath("identity")
log.info(f"Writing key")
keyfile.write_text(key_contents)
keyfile.chmod(0o600)
log.info(f"Checking supplied key for validity...")
rc = subprocess.run(["/usr/bin/ssh-keygen", "-y", "-f", keyfile.as_posix(), "-P", ""], stdout = subprocess.PIPE)
if not rc.returncode == 0:
log.critical(f"Supplied key is invalid.")
if input_default_bool("Do you want to try again? [y/n]", True):
reuse_ssh_key(ssh_dir)
else:
print("Public key is:")
print("-----------------------------------------------------")
public = rc.stdout.decode().strip() + f" cloner-deploy-key-{data['cloner_project_name']}"
print(public)
print("-----------------------------------------------------")
log.info(f"Writing new public key...")
keyfile_pub.write_text(public)
def solve_authorization(conf_dir: Path):
decision: str = pyip.inputMenu(
["Create ssh key", "Reuse ssh key", "Keep unchanged / unauthorized / usernames access only"],
numbered = True)
log.info(f"Preparing auth file..")
ssh_dir = conf_dir.joinpath("auth", "ssh")
if not ssh_dir.exists():
ssh_dir.mkdir(mode = 0o700, parents = True)
if decision.startswith("Create"):
create_ssh_key(ssh_dir)
elif decision.startswith("Reuse"):
reuse_ssh_key(ssh_dir)
def main() -> int:
check_privileges()
parse_args()
# determine debug
debug = pyip.inputYesNo("Enable verbose logging? [y/N]", default = "no", blank = True)
if debug == "yes":
log.info("Setting verbose logging")
log.setLevel(logging.DEBUG)
# defaults
data["cloner_project_name"] = ""
data["cloner_repo_url"] = ""
data["cloner_interval"] = 5
data["cloner_submodules"] = False
data["cloner_submodule_depth"] = 0
data["detector"] = True
data["detector_init"] = True
edit_config = True
# endless loop until finished or canceled
while True:
# query params from user?
if edit_config:
query_repo_info_recursive()
# create dir
project_path = base_dir.joinpath(f"{cloner_prefix}{data['cloner_project_name']}")
global initialized_folder
initialized_folder = True
project_path.mkdir(exist_ok = True)
project_path.joinpath("repos").mkdir(exist_ok = True)
project_path.joinpath("cache").mkdir(exist_ok = True)
config_dir = project_path.joinpath("config")
config_dir.mkdir(exist_ok = True)
gen_config_file(config_dir, **data)
if edit_config:
solve_authorization(config_dir)
# try initializing repos
x = clone_or_fetch(project_path.as_posix(), clone_init = True, detector_init = data["detector_init"])
if x == 0:
# enable repository
project_path.joinpath(".enabled").touch()
return 0
# determine if we want to continue
log.critical(f"Something has failed. Please see log above and decide what to do")
menu_config = "Edit config and start over"
menu_clone_again = "Try cloning again, including base repo"
menu_quit_clean = "Quit & clean"
decision: str = pyip.inputMenu(
[menu_config, menu_clone_again, menu_quit_clean],
numbered = True)
# other stuff
if decision == menu_quit_clean:
log.warning(f"Removing old stuff and quitting")
subprocess.run(["/usr/bin/rm", "-Rvf", project_path.as_posix()])
return 0
if decision == menu_config:
edit_config = True
else:
edit_config = False
# remove caches & repos
log.info(f"Cleaning unwanted stuff")
subprocess.run(["/usr/bin/rm", "-Rvf", project_path.joinpath("repos").as_posix()])
subprocess.run(["/usr/bin/rm", "-Rvf", project_path.joinpath("cache").as_posix()])
log.info(f"Starting over.. ")