refs #2364 commit correct post-install hook

main
Vadim vtroshchinskiy 2025-07-14 11:02:42 +02:00
parent 231fb113a9
commit 34069d82cc
2 changed files with 489 additions and 43 deletions

View File

@ -0,0 +1,489 @@
#!/usr/bin/env python3
"""Script para la instalación del repositorio git"""
import os
import sys
import logging
import subprocess
import pwd
import grp
import pathlib
import socket
import time
import requests
import debconf
class OpengnsysForgejoPostInstall:
"""Opengnsys Forgejo Post-installer"""
def __init__(self):
"""Inicializar clase"""
self.__logger = logging.getLogger("OpengnsysForgejoInstaller")
self.__logger.setLevel(logging.DEBUG)
self.__logger.debug("Initializing")
self.testmode = False
self.base_path = "/opt/opengnsys"
self.ogrepository_base_path = os.path.join(self.base_path, "ogrepository")
self.git_basedir = "base.git"
self.email = "OpenGnsys@opengnsys.com"
self.opengnsys_bin_path = os.path.join(self.base_path, "bin")
self.opengnsys_etc_path = os.path.join(self.base_path, "etc")
self.forgejo_user = "oggit"
self.forgejo_password = "opengnsys"
self.forgejo_organization = "opengnsys"
self.forgejo_port = 3100
self.forgejo_bin_path = os.path.join(self.ogrepository_base_path, "bin")
self.forgejo_exe = os.path.join(self.forgejo_bin_path, "forgejo")
self.forgejo_conf_dir_path = os.path.join(self.ogrepository_base_path, "etc", "forgejo")
self.lfs_dir_path = os.path.join(self.ogrepository_base_path, "oggit", "git-lfs")
self.git_dir_path = os.path.join(self.ogrepository_base_path, "oggit", "git")
self.forgejo_var_dir_path = os.path.join(self.ogrepository_base_path, "var", "lib", "forgejo")
self.forgejo_work_dir_path = os.path.join(self.forgejo_var_dir_path, "work")
self.forgejo_work_custom_dir_path = os.path.join(self.forgejo_work_dir_path, "custom")
self.forgejo_db_dir_path = os.path.join(self.forgejo_var_dir_path, "db")
self.forgejo_data_dir_path = os.path.join(self.forgejo_var_dir_path, "data")
self.forgejo_db_path = os.path.join(self.forgejo_db_dir_path, "forgejo.db")
self.forgejo_log_dir_path = os.path.join(self.ogrepository_base_path, "log", "forgejo")
self.forgejo_ini_file_path = os.path.join(self.forgejo_conf_dir_path, "app.ini")
self.ssh_group = "oggit"
self.ssh_user = "oggit"
self.ssh_uid = -1
self.ssh_gid = -1
self.ssh_homedir = None
self.temp_dir = None
self.script_path = os.path.realpath(os.path.dirname(__file__))
# Where we look for forgejo-app.ini and similar templates.
self.template_path = self.script_path
def _apply_configuration(self):
"""Do anything that requires recalculating variables based on other variables.
This ensures any changes made based on debconf configs are acted on.
"""
self.set_ssh_user_group(self.ssh_user, self.ssh_group)
def set_ssh_user_group(self, username, groupname):
"""
Ensures that the specified SSH user and group exist on the system, creating them if necessary.
Args:
username (str): The name of the SSH user to ensure exists.
groupname (str): The name of the SSH group to ensure exists.
Side Effects:
- Sets instance attributes: ssh_group, ssh_user, ssh_gid, ssh_uid, ssh_homedir.
- Logs actions and results using the instance logger.
- May create a new system group and/or user if they do not already exist.
Raises:
subprocess.CalledProcessError: If group or user creation commands fail.
"""
self.ssh_group = groupname
self.ssh_user = username
try:
self.ssh_gid = grp.getgrnam(self.ssh_group).gr_gid
self.__logger.info("Group %s exists with gid %i", self.ssh_group, self.ssh_gid)
except KeyError:
self.__logger.info("Need to create group %s", self.ssh_group)
subprocess.run(["/usr/sbin/groupadd", "--system", self.ssh_group], check=True)
self.ssh_gid = grp.getgrnam(groupname).gr_gid
try:
self.ssh_uid = pwd.getpwnam(self.ssh_user).pw_uid
self.__logger.info("User %s exists with gid %i", self.ssh_user, self.ssh_uid)
except KeyError:
self.__logger.info("Need to create user %s", self.ssh_user)
subprocess.run(["/usr/sbin/useradd", "--gid", str(self.ssh_gid), "-m", "--system", self.ssh_user], check=True)
self.ssh_uid = pwd.getpwnam(username).pw_uid
self.ssh_homedir = pwd.getpwnam(username).pw_dir
def _recursive_chown(self, path, ouid, ogid):
"""
Recursively change the owner and group of a directory and all its files.
Args:
path (str): The root directory path to start changing ownership.
ouid (int): The user ID to set as the owner.
ogid (int): The group ID to set as the group.
Raises:
OSError: If changing ownership fails for any file or directory.
"""
for dirpath, _, filenames in os.walk(path):
os.chown(dirpath, uid=ouid, gid=ogid)
for filename in filenames:
os.chown(os.path.join(dirpath, filename), uid=ouid, gid=ogid)
def _wait_for_port(self, host, port):
self.__logger.info("Waiting for %s:%i to be up", host, port)
timeout = 60
start_time = time.time()
ready = False
while not ready and (time.time() - start_time) < 60:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
s.connect((host, port))
ready = True
s.close()
except TimeoutError:
self.__logger.debug("Timed out, no connection yet.")
except OSError as oserr:
self.__logger.debug("%s, no connection yet. %.1f seconds left.", oserr.strerror, timeout - (time.time() - start_time))
time.sleep(0.1)
if ready:
self.__logger.info("Connection established.")
else:
self.__logger.error("Timed out waiting for connection!")
raise TimeoutError("Timed out waiting for connection!")
def _install_template(self, template, destination, keysvalues):
self.__logger.info("Writing template %s into %s", template, destination)
data = ""
with open(template, "r", encoding="utf-8") as template_file:
data = template_file.read()
for key in keysvalues.keys():
if isinstance(keysvalues[key], int):
data = data.replace("{" + key + "}", str(keysvalues[key]))
else:
data = data.replace("{" + key + "}", keysvalues[key])
with open(destination, "w+", encoding="utf-8") as out_file:
out_file.write(data)
def _runcmd(self, cmd):
self.__logger.debug("Running: %s", cmd)
ret = subprocess.run(cmd, check=True,capture_output=True, encoding='utf-8')
return ret.stdout.strip()
def _get_forgejo_data(self):
data = {
"forgejo_user" : self.ssh_user,
"forgejo_group" : self.ssh_group,
"forgejo_port" : str(self.forgejo_port),
"forgejo_bin" : self.forgejo_exe,
"forgejo_app_ini" : self.forgejo_ini_file_path,
"forgejo_work_path" : self.forgejo_work_dir_path,
"forgejo_data_path" : self.forgejo_data_dir_path,
"forgejo_db_path" : self.forgejo_db_path,
"forgejo_repository_root" : self.git_dir_path,
"forgejo_lfs_path" : self.lfs_dir_path,
"forgejo_log_path" : self.forgejo_log_dir_path,
"forgejo_hostname" : self._runcmd("hostname"),
"forgejo_lfs_jwt_secret" : self._runcmd([self.forgejo_exe,"generate", "secret", "LFS_JWT_SECRET"]),
"forgejo_jwt_secret" : self._runcmd([self.forgejo_exe,"generate", "secret", "JWT_SECRET"]),
"forgejo_internal_token" : self._runcmd([self.forgejo_exe,"generate", "secret", "INTERNAL_TOKEN"]),
"forgejo_secret_key" : self._runcmd([self.forgejo_exe,"generate", "secret", "SECRET_KEY"])
}
return data
def run_forge_cmd(self, args, ignore_errors = None):
"""
Executes a Forgejo command with the specified arguments and handles errors.
Args:
args (list): List of command-line arguments to pass to the Forgejo executable.
ignore_errors (list, optional): List of error message substrings. If any of these are found in the command's stderr output, the error is ignored and the command's stdout is returned. Defaults to an empty list.
Returns:
str: The standard output of the command if it succeeds or if an ignored error occurs.
Raises:
RuntimeError: If the command fails and the error is not in the ignore_errors list.
Logs:
- Info: The command being run.
- Error: If the command fails, logs the command, return code, stdout, and stderr.
- Info: If an error is ignored, logs that the error is being ignored.
"""
if ignore_errors is None:
ignore_errors = []
cmd = [self.forgejo_exe, "--config", self.forgejo_ini_file_path] + args
self.__logger.info("Running command: %s", cmd)
ret = subprocess.run(cmd, check=False, capture_output=True, encoding='utf-8', user=self.ssh_user)
if ret.returncode == 0:
return ret.stdout.strip()
else:
self.__logger.error("Failed to run command: %s, return code %i", cmd, ret.returncode)
self.__logger.error("stdout: %s", ret.stdout.strip())
self.__logger.error("stderr: %s", ret.stderr.strip())
for err in ignore_errors:
if err in ret.stderr:
self.__logger.info("Ignoring error, it's in the ignore list")
return ret.stdout.strip()
raise RuntimeError("Failed to run necessary command")
def configure_forgejo(self):
"""
Configures the Forgejo service for OpenGnsys.
This method performs the following steps:
1. Retrieves Forgejo configuration data.
2. Creates necessary directories for configuration, data, logs, and workspaces.
3. Sets ownership of key directories to the SSH user and group.
4. Installs configuration and systemd service templates.
5. Reloads systemd, enables, and restarts the Forgejo service.
6. Waits for the Forgejo service to become available on the configured port.
7. Runs Forgejo administrative commands to:
- Migrate the database.
- Perform a health check.
- Create the admin user (if not already present).
- Generate an access token for the admin user (if not already present).
8. Stores the generated access token in a configuration file, or retains the old token if generation fails.
Raises:
RuntimeError: If any required Forgejo administrative command fails and the error is not in the ignore list.
"""
# Make sure any custom config from debconf is accounted for
self._apply_configuration()
data = self._get_forgejo_data()
self.__logger.debug("Creating directories")
pathlib.Path(self.opengnsys_etc_path).mkdir(parents=True, exist_ok=True)
pathlib.Path(self.forgejo_conf_dir_path).mkdir(parents=True, exist_ok=True)
pathlib.Path(self.git_dir_path).mkdir(parents=True, exist_ok=True)
pathlib.Path(self.lfs_dir_path).mkdir(parents=True, exist_ok=True)
pathlib.Path(self.forgejo_work_dir_path).mkdir(parents=True, exist_ok=True)
pathlib.Path(self.forgejo_work_custom_dir_path).mkdir(parents=True, exist_ok=True)
pathlib.Path(self.forgejo_data_dir_path).mkdir(parents=True, exist_ok=True)
pathlib.Path(self.forgejo_db_dir_path).mkdir(parents=True, exist_ok=True)
pathlib.Path(self.forgejo_log_dir_path).mkdir(parents=True, exist_ok=True)
os.chown(self.lfs_dir_path, self.ssh_uid, self.ssh_gid)
os.chown(self.git_dir_path, self.ssh_uid, self.ssh_gid)
os.chown(self.forgejo_data_dir_path, self.ssh_uid, self.ssh_gid)
os.chown(self.forgejo_work_dir_path, self.ssh_uid, self.ssh_gid)
os.chown(self.forgejo_db_dir_path, self.ssh_uid, self.ssh_gid)
os.chown(self.forgejo_log_dir_path, self.ssh_uid, self.ssh_gid)
self._install_template(os.path.join(self.template_path, "forgejo-app.ini"), self.forgejo_ini_file_path, data)
self._install_template(os.path.join(self.template_path, "opengnsys-forgejo.service"), "/etc/systemd/system/opengnsys-forgejo.service", data)
self.__logger.debug("Reloading systemd and starting service")
subprocess.run(["systemctl", "daemon-reload"], check=True)
subprocess.run(["systemctl", "enable", "opengnsys-forgejo"], check=True)
subprocess.run(["systemctl", "restart", "opengnsys-forgejo"], check=True)
self.__logger.info("Waiting for forgejo to start")
self._wait_for_port("localhost", self.forgejo_port)
self.__logger.info("Configuring forgejo")
self.run_forge_cmd(["migrate"])
self.run_forge_cmd(["admin", "doctor", "check"])
self.run_forge_cmd(["admin", "user", "create", "--username", self.forgejo_user, "--password", self.forgejo_password, "--email", self.email], ignore_errors=["user already exists"])
token = self.run_forge_cmd(["admin", "user", "generate-access-token", "--username", self.forgejo_user, "-t", "gitapi", "--scopes", "all", "--raw"], ignore_errors = ["access token name has been used already"])
if token:
with open(os.path.join(self.base_path, "etc", "ogGitApiToken.cfg"), "w+", encoding='utf-8') as token_file:
token_file.write(token)
else:
self.__logger.info("Keeping the old token")
def add_forgejo_repo(self, repository_name, description = ""):
"""
Create a new repository in the local Forgejo instance.
This method sends a POST request to the Forgejo API to create a new repository
with the specified name and optional description. The API token is read from
the configuration file 'ogGitApiToken.cfg'.
Args:
repository_name (str): The name of the repository to create.
description (str, optional): A description for the repository. Defaults to "".
Raises:
requests.RequestException: If the HTTP request to Forgejo fails.
Logs:
Information about the repository creation request and its response status.
"""
token = ""
with open(os.path.join(self.base_path, "etc", "ogGitApiToken.cfg"), "r", encoding='utf-8') as token_file:
token = token_file.read().strip()
self.__logger.info("Adding repository %s for Forgejo", repository_name)
r = requests.post(
f"http://localhost:{self.forgejo_port}/api/v1/user/repos",
json={
"auto_init" : False,
"default_branch" : "main",
"description" : description,
"name" : repository_name,
"private" : False
}, headers={
'Authorization' : f"token {token}"
},
timeout = 60
)
self.__logger.info("Request status was %i, content %s", r.status_code, r.content)
def add_forgejo_sshkey(self, pubkey, description = ""):
"""
Adds an SSH public key to the Forgejo server for the current user.
This method reads an API token from the configuration file and sends a POST request
to the Forgejo API to add the specified SSH public key with an optional description.
Args:
pubkey (str): The SSH public key to be added.
description (str, optional): A description or title for the SSH key. Defaults to "".
Returns:
tuple: A tuple containing the HTTP status code (int) and the response content (str).
Raises:
requests.RequestException: If the HTTP request fails.
FileNotFoundError: If the API token configuration file does not exist.
Exception: For any other unexpected errors.
"""
token = ""
with open(os.path.join(self.base_path, "etc", "ogGitApiToken.cfg"), "r", encoding='utf-8') as token_file:
token = token_file.read().strip()
self.__logger.info("Adding SSH key to Forgejo: %s (%s)", pubkey, description)
r = requests.post(
f"http://localhost:{self.forgejo_port}/api/v1/user/keys",
json={
"key" : pubkey,
"read_only" : False,
"title" : description
}, headers={
'Authorization' : f"token {token}"
},
timeout = 60
)
self.__logger.info("Request status was %i, content %s", r.status_code, r.content)
return r.status_code, r.content.decode('utf-8')
def add_forgejo_organization(self, pubkey, description = ""):
"""
Adds an SSH public key to a Forgejo instance for the current user.
This method reads an API token from a configuration file, then sends a POST request
to the Forgejo API to add the provided SSH public key. The key can be given a description,
which is used as its title in Forgejo.
Args:
pubkey (str): The SSH public key to add.
description (str, optional): A description or title for the key. Defaults to "".
Logs:
- Information about the key being added.
- The status code and content of the API response.
Raises:
requests.RequestException: If the HTTP request fails or times out.
FileNotFoundError: If the API token configuration file does not exist.
Exception: For other unexpected errors.
"""
token = ""
with open(os.path.join(self.base_path, "etc", "ogGitApiToken.cfg"), "r", encoding='utf-8') as token_file:
token = token_file.read().strip()
self.__logger.info("Adding SSH key to Forgejo: %s", pubkey)
r = requests.post(
f"http://localhost:{self.forgejo_port}/api/v1/user/keys",
json={
"key" : pubkey,
"read_only" : False,
"title" : description
}, headers={
'Authorization' : f"token {token}"
},
timeout = 60
)
self.__logger.info("Request status was %i, content %s", r.status_code, r.content)
if __name__ == '__main__':
sys.stdout.reconfigure(encoding='utf-8')
opengnsys_log_dir = "/opt/opengnsys/log"
logger = logging.getLogger(__package__)
logger.setLevel(logging.DEBUG)
streamLog = logging.StreamHandler()
streamLog.setLevel(logging.INFO)
pathlib.Path(opengnsys_log_dir).mkdir(parents=True, exist_ok=True)
logFilePath = f"{opengnsys_log_dir}/git_installer.log"
fileLog = logging.FileHandler(logFilePath)
fileLog.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(name)24s - [%(levelname)5s] - %(message)s')
streamLog.setFormatter(formatter)
fileLog.setFormatter(formatter)
logger.addHandler(streamLog)
logger.addHandler(fileLog)
logger.info("Running post-install script")
installer=OpengnsysForgejoPostInstall()
logger.debug("Obtaining configuration from debconf")
with debconf.Debconf(run_frontend=True) as db:
def dbget(value):
ret = db.get(value)
logger.debug("Retrieved %s from debconf: %s", value, ret)
return ret
installer.forgejo_organization = dbget('opengnsys/forgejo_organization')
installer.forgejo_user = dbget('opengnsys/forgejo_user')
installer.forgejo_password = dbget('opengnsys/forgejo_password')
installer.email = dbget('opengnsys/forgejo_email')
installer.forgejo_port = int(dbget('opengnsys/forgejo_port'))
# Templates get installed here
installer.template_path = "/usr/share/opengnsys-forgejo/"
installer.configure_forgejo()
sys.exit(0)

View File

@ -1,43 +0,0 @@
#!/usr/bin/env python3
data = {
"forgejo_user" : self.ssh_user,
"forgejo_group" : self.ssh_group,
"forgejo_port" : str(self.forgejo_port),
"forgejo_bin" : bin_path,
"forgejo_app_ini" : conf_path,
"forgejo_work_path" : forgejo_work_dir_path,
"forgejo_data_path" : forgejo_data_dir_path,
"forgejo_db_path" : forgejo_db_path,
"forgejo_repository_root" : git_dir_path,
"forgejo_lfs_path" : lfs_dir_path,
"forgejo_log_path" : forgejo_log_dir_path,
"forgejo_hostname" : _runcmd("hostname"),
"forgejo_lfs_jwt_secret" : _runcmd([bin_path,"generate", "secret", "LFS_JWT_SECRET"]),
"forgejo_jwt_secret" : _runcmd([bin_path,"generate", "secret", "JWT_SECRET"]),
"forgejo_internal_token" : _runcmd([bin_path,"generate", "secret", "INTERNAL_TOKEN"]),
"forgejo_secret_key" : _runcmd([bin_path,"generate", "secret", "SECRET_KEY"])
}
ini_template = "/usr/share/opengnsys-forgejo/forgejo-app.ini"
def _install_template(self, template, destination, keysvalues):
data = ""
with open(template, "r", encoding="utf-8") as template_file:
data = template_file.read()
for key in keysvalues.keys():
if isinstance(keysvalues[key], int):
data = data.replace("{" + key + "}", str(keysvalues[key]))
else:
data = data.replace("{" + key + "}", keysvalues[key])
with open(destination, "w+", encoding="utf-8") as out_file:
out_file.write(data)
_install_template(os.path.join(self.script_path, "forgejo-app.ini"), conf_path, data)