+.idea

+venv

+

+__pycache__

+3.11

+#! /usr/bin/env sh

+exec python3 "$(dirname "$0")/aio.py" "$@"

+#!/usr/bin/env python3

+

+try:

+ import rich

+ import jsonschema

+ import cryptography

+except ImportError:

+ print("Some necessary crupest can't be imported. Please run `pip install -r requirements.txt` to install them.")

+ exit(1)

+

+from os.path import *

+import argparse

+import subprocess

+from rich.prompt import Confirm

+from crupest.install_docker import *

+from crupest.path import *

+from crupest.nginx import *

+from crupest.config import *

+from crupest.check import *

+from crupest.backup import *

+from crupest.download_tools import *

+from crupest.test import *

+from crupest.dns import *

+from crupest.setup import *

+

+from crupest.tui import console

+

+

+parser = argparse.ArgumentParser(

+ description="Crupest server all-in-one setup script. Have fun play with it!")

+parser.add_argument("--no-hello", action="store_true",

+ default=False, help="Do not print hello message.")

+parser.add_argument("--no-bye-bye", action="store_true",

+ default=False, help="Do not print bye-bye message.")

+

+parser.add_argument("--no-check-python-version", action="store_true",

+ default=False, help="Do not check python version.")

+parser.add_argument("--no-check-system", action="store_true",

+ default=False, help="Do not check system type.")

+parser.add_argument("-y", "--yes", action="store_true",

+ default=False, help="Yes to all confirmation.")

+

+subparsers = parser.add_subparsers(dest="action")

+

+setup_parser = subparsers.add_parser(

+ "setup", help="Do everything necessary to setup the server.")

+

+print_path_parser = subparsers.add_parser(

+ "print-path", help="Print the paths of all related files and dirs.")

+

+download_tools_parser = subparsers.add_parser(

+ "download-tools", help="Download some extra tools to manage the server.")

+

+list_domain_parser = subparsers.add_parser(

+ "list-domain", help="Misc things about domains.")

+

+nginx_parser = subparsers.add_parser(

+ "nginx", help="Generate nginx config.")

+

+certbot_parser = subparsers.add_parser(

+ "certbot", help="Get some common certbot commands.")

+

+certbot_command_group = certbot_parser.add_mutually_exclusive_group()

+

+certbot_command_group.add_argument(

+ "-C", "--create", action="store_true", default=False, help="Only print the command for 'create' action.")

+certbot_command_group.add_argument(

+ "-E", "--expand", action="store_true", default=False, help="Only print the command for 'expand' action.")

+certbot_command_group.add_argument(

+ "-R", "--renew", action="store_true", default=False, help="Only print the command for 'renew' action.")

+

+certbot_parser.add_argument(

+ "-t", "--test", action="store_true", default=False, help="Make the commands for test use.")

+

+clear_parser = subparsers.add_parser(

+ "clear", help="Delete existing data so you can make a fresh start.")

+clear_parser.add_argument("-D", "--include-data-dir", action="store_true",

+ default=False, help="Also delete the data directory.")

+

+install_docker_parser = subparsers.add_parser(

+ "install-docker", help="Install docker and docker-compose.")

+

+backup_parser = subparsers.add_parser(

+ "backup", help="Backup related things."

+)

+

+backup_subparsers = backup_parser.add_subparsers(dest="backup_action")

+backup_restore_parser = backup_subparsers.add_parser(

+ "restore", help="Restore data from url.")

+backup_restore_parser.add_argument(

+ "restore_url", help="Restore archive url. Can be local path or http/https.")

+backup_backup_parser = backup_subparsers.add_parser(

+ "backup", help="Backup data to specified path.")

+backup_backup_parser.add_argument(

+ "backup_path", nargs="?", help="Backup path. Can be empty for a timestamp as name. Must be local path.")

+

+docker_parser = subparsers.add_parser("docker", help="Docker related things.")

+docker_subparsers = docker_parser.add_subparsers(dest="docker_action")

+docker_subparsers.add_parser("up", help="Run docker compose up -d.")

+docker_subparsers.add_parser("down", help="Run docker compose down.")

+docker_subparsers.add_parser(

+ "prune", help="Run docker system prune -a -f.")

+

+test_parser = subparsers.add_parser("test", help="Test things.")

+test_parser.add_argument(

+ "test_action", help="Test action.", choices=["crupest-api"])

+

+dns_parser = subparsers.add_parser("dns", help="Generate dns zone.")

+

+dns_parser.add_argument("-i", "--ip", help="IP address of the server.")

+

+git_update_parser = subparsers.add_parser(

+ "git-update", help="Update git submodules.")

+

+update_blog_parser = subparsers.add_parser(

+ "update-blog", help="Update and regenerate blog.")

+

+up_parser = subparsers.add_parser(

+ "up", help="Do something necessary and then docker compose up.")

+

+down_parser = subparsers.add_parser(

+ "down", help="Do something necessary and then docker compose down.")

+

+args = parser.parse_args()

+

+if args.yes:

+ old_ask = Confirm.ask

+

+ def new_ask(prompt, *args, console=console, default=None, **kwargs):

+ default_text = ""

+ if default is not None:

+ default_text = "(y)" if default else "(n)"

+ text = f"[prompt]{prompt}[/] [prompt.choices]\\[y/n][/] [prompt.default]{default_text}[/]"

+ console.print(text)

+ return True

+

+ Confirm.ask = new_ask

+

+if (args.action == "certbot" and (args.create or args.renew or args.expand)) or (args.action == "dns" and args.ip is not None):

+ args.no_hello = True

+ args.no_bye_bye = True

+

+

+if not args.no_check_python_version:

+ if check_python_version():

+ console.print("This script works well on python 3.10. Otherwise you may encounter some problems. But I would like to improve some rational compatibility.", style="yellow")

+

+if not args.no_check_system:

+ if not check_ubuntu():

+ console.print("This script works well on Ubuntu 22.04. Otherwise you may encounter some problems. But I would like to improve some rational compatibility.", style="yellow")

+

+

+if not args.no_hello:

+ console.print("Nice to see you! :waving_hand:", style="cyan")

+

+

+def check_domain_is_defined():

+ try:

+ return get_domain()

+ except Exception as e:

+ console.print(e.args[0], style="red")

+ raise e

+

+

+def git_update():

+ def do_it():

+ subprocess.run(["git", "pull"], check=True)

+ run_in_project_dir(do_it)

+

+

+def update_blog():

+ def do_it():

+ subprocess.run(["docker", "compose", "exec",

+ "crupest-blog", "/scripts/update.bash"], check=True)

+ run_in_project_dir(do_it)

+

+

+def docker_compose_up():

+ def do_docker_compose_up():

+ subprocess.run(["docker", "compose", "up", "-d"], check=True)

+ run_in_dir(project_abs_path, do_docker_compose_up)

+

+

+def docker_compose_down():

+ def do_docker_compose_down():

+ subprocess.run(

+ ["docker", "compose", "down"], check=True)

+ run_in_dir(project_abs_path, do_docker_compose_down)

+

+

+action = args.action

+

+

+def run():

+ match action:

+ case "install-docker":

+ install_docker()

+ console.print(

+ "Succeeded to install docker. Please re-login to take effect.", style="green")

+

+ case "docker":

+ docker_action = args.docker_action

+

+ match docker_action:

+ case "up":

+ docker_compose_up()

+ case "down":

+ docker_compose_down()

+ case "prune":

+ to_do = Confirm.ask(

+ "[yellow]Are you sure to prune docker?[/]", console=console)

+ if to_do:

+ subprocess.run(

+ ["docker", "system", "prune", "-a", "-f"], check=True)

+ case _:

+ raise ValueError("Unknown docker action.")

+

+ case "backup":

+ backup_action = args.backup_action

+ match backup_action:

+ case "backup":

+ backup_backup(args.backup_path, console)

+ console.print("Succeeded to restore data.", style="green")

+ case "restore":

+ backup_restore(args.restore_path, console)

+ console.print("Succeeded to backup data.", style="green")

+

+ case 'print-path':

+ console.print("Project path =", project_dir)

+ console.print("Project absolute path =", project_abs_path)

+ console.print("Data path =", data_dir)

+

+ case "download-tools":

+ download_tools(console)

+

+ case "list-domain":

+ domain = check_domain_is_defined()

+ domains = list_domains(domain)

+ for domain in domains:

+ console.print(domain)

+

+ case "nginx":

+ raise Exception("This command is deprecated.")

+

+ case "certbot":

+ domain = check_domain_is_defined()

+ is_test = args.test

+ if args.create:

+ console.print(certbot_command_gen(domain, "create",

+ test=is_test), soft_wrap=True, highlight=False)

+ elif args.expand:

+ console.print(certbot_command_gen(domain, "expand",

+ test=is_test), soft_wrap=True, highlight=False)

+ elif args.renew:

+ console.print(certbot_command_gen(domain, "renew",

+ test=is_test), soft_wrap=True, highlight=False)

+ else:

+ console.print(

+ "Here is some commands you can use to do certbot related work.")

+ if is_test:

+ console.print(

+ "Note you specified --test, so the commands are for test use.", style="yellow")

+ console.print(

+ "To create certs for init (standalone):", style="cyan")

+ console.print(certbot_command_gen(

+ domain, 'create', test=is_test), soft_wrap=True)

+ console.print("To expand certs (nginx):", style="cyan")

+ console.print(certbot_command_gen(

+ domain, 'create', test=is_test), soft_wrap=True)

+ console.print(

+ "To renew certs previously created (nginx):", style="cyan")

+ console.print(certbot_command_gen(

+ domain, 'renew', test=is_test), soft_wrap=True)

+ case "test":

+ match args.test_action:

+ case "crupest-api":

+ test_crupest_api(console)

+ case _:

+ console.print("Test action invalid.", style="red")

+

+ case "dns":

+ domain = check_domain_is_defined()

+ if domain is not None:

+ if args.ip is None:

+ ip = Prompt.ask(

+ "Please enter your server ip", console=console)

+ else:

+ ip = args.ip

+ console.print(generate_dns_zone_with_dkim(

+ domain, ip), soft_wrap=True, highlight=False)

+

+ case "git-update":

+ git_update()

+

+ case "update-blog":

+ update_blog()

+

+ case "up":

+ git_update()

+ template_generate(console)

+ docker_compose_up()

+

+ case "down":

+ docker_compose_down()

+

+ case "clear":

+ clear(console, args.include_data_dir)

+

+ case _:

+ template_generate(console)

+ if Confirm.ask(

+ "By the way, would you like to download some scripts to do some extra setup like creating email user?", console=console, default=True):

+ download_tools(console)

+

+

+run()

+

+if not args.no_bye_bye:

+ console.print(":beers: All done! Bye bye!", style="green")

+import sys

+

+

+class CruInitError(Exception):

+ pass

+

+def check_python_version(required_version=(3, 11)):

+ if sys.version_info < required_version:

+ raise CruInitError(f"Python version must be >= {required_version}!")

+

+

+check_python_version()

+from collections.abc import Callable

+from dataclasses import dataclass

+from types import NoneType

+from typing import Any

+from copy import deepcopy

+

+

+@dataclass

+class CruAttr:

+ name: str

+ value: Any

+ description: str

+

+

+@dataclass

+class CruAttrDef:

+ name: str

+ default_description: str

+ allow_types: None | set[type]

+ allow_none: bool

+ default_value: Any

+ transformer: Callable[[Any], Any] | None

+ validator: Callable[[Any], None]

+

+ def __init__(self, name: str, default_description: str, *,

+ allow_types: list[type] | type | None, allow_none: bool, default_value: Any = None,

+ transformer: Callable[[Any], Any] | None = None,

+ validator: Callable[[Any], None] | None = None) -> None:

+ self.name = name

+ self.default_description = default_description

+ self.default_value = default_value

+ #TODO: CONTINUE TOMORROW

+ if allow_types is None:

+ allow_types = []

+ elif isinstance(allow_types, type):

+ allow_types = [allow_types]

+ else:

+ for t in allow_types:

+ if not isinstance(t, type):

+ raise TypeError(f"Invalid value of python type : {t}")

+ self.allow_types = set(filter(lambda tt: tt is not NoneType, allow_types))

+ self.allow_none = allow_none

+ self.transformer = transformer

+ self.validator = validator

+ self.default_value = self.transform_and_validate(self.default_value)

+ self.default_value = deepcopy(self.default_value)

+

+ def transform(self, value: Any) -> Any:

+ if self.transformer is not None:

+ return self.transformer(value)

+ return value

+

+ def validate(self, value: Any, /, override_allow_none: bool | None = None) -> None:

+ allow_none = override_allow_none if override_allow_none is not None else self.allow_none

+ if value is None and not allow_none:

+ raise TypeError(f"None is not allowed!")

+ if len(self.allow_types) != 0 and type(value) not in self.allow_types:

+ raise TypeError(f"Type of {value} is not allowed!")

+ if self.validator is not None:

+ return self.validator(value)

+ return None

+

+ def transform_and_validate(self, value: Any, /, override_allow_none: bool | None = None) -> Any:

+ value = self.transform(value)

+ self.validate(value, override_allow_none)

+ return value

+

+ def make(self, value: Any, description: None | str = None) -> CruAttr:

+ value = self.transform_and_validate(value)

+ return CruAttr(self.name, value if value is not None else deepcopy(self.default_value),

+ description if description is not None else self.default_description)

+

+

+class CruAttrDefRegistry:

+

+ def __init__(self) -> None:

+ self._def_list = []

+

+ @property

+ def items(self) -> list[CruAttrDef]:

+ return self._def_list

+

+ def register(self, def_: CruAttrDef):

+ for i in self._def_list:

+ if i.name == def_.name:

+ raise ValueError(f"Attribute {def_.name} already exists!")

+ self._def_list.append(def_)

+

+ def register_with(self, name: str, default_description: str, *,

+ allow_types: list[type] | type | None, allow_none: bool,

+ default_value: Any = None,

+ transformer: Callable[[Any], Any] | None = None,

+ validator: Callable[[Any], None] | None = None

+ ) -> CruAttrDef:

+ def_ = CruAttrDef(name, default_description, default_value=default_value, allow_types=allow_types,

+ allow_none=allow_none, transformer=transformer, validator=validator)

+ self.register(def_)

+ return def_

+

+ def register_required(self, name: str, default_description: str, *,

+ allow_types: list[type] | type | None,

+ default_value: Any = None,

+ transformer: Callable[[Any], Any] | None = None,

+ validator: Callable[[Any], None] | None = None

+ ) -> CruAttrDef:

+ return self.register_with(name, default_description, default_value=default_value, allow_types=allow_types,

+ allow_none=False, transformer=transformer, validator=validator)

+

+ def register_optional(self, name: str, default_description: str, *,

+ allow_types: list[type] | type | None,

+ default_value: Any = None,

+ transformer: Callable[[Any], Any] | None = None,

+ validator: Callable[[Any], None] | None = None

+ ) -> CruAttrDef:

+ return self.register_with(name, default_description, default_value=default_value, allow_types=allow_types,

+ allow_none=True, transformer=transformer, validator=validator)

+

+ def get_item_optional(self, name: str) -> CruAttrDef | None:

+ for i in self._def_list:

+ if i.name == name:

+ return i

+ return None

+

+ def __getitem__(self, item) -> CruAttrDef | None:

+ return self.get_item_optional(item)

+from typing import Any, TypeVar, Generic

+

+from .excp import CruInternalLogicError

+from .value import ValueType, ValueGenerator, ValidationError

+

+T = TypeVar("T")

+

+

+class ConfigItem(Generic[T]):

+ OptionalValueGenerator = ValueGenerator[T, []] | None

+

+ def __init__(self, name: str, description: str, value_type: ValueType[T], value: T | None, default_value: T, *,

+ value_generator: OptionalValueGenerator = None) -> None:

+ self._name = name

+ self._description = description

+ self._value_type = value_type

+ self._default_value = default_value

+ self._value_generator = value_generator

+ self._value: T | None = value

+

+ @property

+ def name(self) -> str:

+ return self._name

+

+ @property

+ def description(self) -> str:

+ return self._description

+

+ @property

+ def value_type(self) -> ValueType[T]:

+ return self._value_type

+

+ @property

+ def default_value(self) -> T:

+ return self._default_value

+

+ @property

+ def is_default(self) -> bool:

+ return self._value is None

+

+ @property

+ def is_set(self) -> bool:

+ return not self.is_default

+

+ @property

+ def value(self) -> T:

+ return self._value or self._default_value

+

+ def set_value(self, v: T | str, /, allow_convert_from_str=False):

+ if allow_convert_from_str:

+ self._value = self.value_type.check_value(v)

+ else:

+ self._value = self.value_type.check_value_or_try_convert_from_str(v)

+

+ @value.setter

+ def value(self, v: T) -> None:

+ self.set_value(v)

+

+ @property

+ def value_generator(self) -> OptionalValueGenerator:

+ return self._value_generator

+

+ def generate_value(self, allow_interactive=False) -> T | None:

+ if self.value_generator is None: return None

+ if self.value_generator.interactive and not allow_interactive:

+ return None

+ else:

+ v = self.generate_value()

+ try:

+ self.value_type.check_value(v)

+ return v

+ except ValidationError as e:

+ raise CruInternalLogicError("Config value generator returns invalid value.", name=self.name, inner=e)

+

+ def copy(self) -> "ConfigItem":

+ return ConfigItem(self.name, self.description, self.value_type,

+ self._value.copy() if self._value is not None else None, self._default_value.copy(),

+ value_generator=self.value_generator)

+

+

+class Configuration:

+ def __init__(self, items: None | list[ConfigItem] = None) -> None:

+ self._items: list[ConfigItem] = items or []

+

+ @property

+ def items(self) -> list[ConfigItem]:

+ return self._items

+

+ @property

+ def item_map(self) -> dict[str, ConfigItem]:

+ return {i.name: i for i in self.items}

+

+ def get_optional_item(self, name: str) -> ConfigItem | None:

+ for i in self.items:

+ if i.name == name:

+ return i

+ return None

+

+ def clear(self) -> None:

+ self._items.clear()

+

+ def has_item(self, name: str) -> bool:

+ return self.get_optional_item(name) is not None

+

+ def add_item(self, item: ConfigItem):

+ i = self.get_optional_item(item.name)

+ if i is not None:

+ raise CruInternalLogicError("Config item of the name already exists.", name=item.name)

+ self.items.append(item)

+ return item

+

+ def set_value(self, name: str, v: Any, /, allow_convert_from_str=False):

+ i = self.get_optional_item(name)

+ if i is None:

+ raise CruInternalLogicError("No config item of the name. Can't set value.", name=name)

+ i.set_value(v, allow_convert_from_str)

+

+ def copy(self) -> "Configuration":

+ return Configuration([i.copy() for i in self.items])

+

+ def __getitem__(self, name: str) -> ConfigItem:

+ i = self.get_optional_item(name)

+ if i is not None:

+ return i

+ raise CruInternalLogicError('No config item of the name.', name=name)

+

+ def __contains__(self, name: str):

+ return self.has_item(name)

+from collections.abc import Callable

+from dataclasses import dataclass

+from types import NoneType

+from typing import Any

+

+from cru.attr import CruAttrDefRegistry

+

+CRU_EXCEPTION_ATTR_DEF_REGISTRY = CruAttrDefRegistry()

+

+

+class CruException(Exception):

+ @staticmethod

+ def transform_inner(inner: Exception | list[Exception] | None):

+ if inner is None:

+ return None

+ if isinstance(inner, Exception):

+ return [inner]

+ if isinstance(inner, list):

+ return inner

+

+ @staticmethod

+ def validate_inner(inner: list[Exception]):

+ for i in inner:

+ if not isinstance(i, Exception):

+ raise TypeError(f"Invalid inner exception: {i}")

+

+ MESSAGE_DEF = CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_required("message", "Message describing the exception.",

+ allow_types=str, default_value="")

+ INNER_DEF = CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_required("inner", "Inner exception.",

+ allow_types=list, default_value=[],

+ transformer=transform_inner, validator=validate_inner)

+ INTERNAL_DEF = CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_required("internal",

+ "True if the exception is caused by wrong internal logic. False if it is caused by user's wrong input.",

+ allow_types=bool, default_value=False)

+ CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_optional("name", "Name of the object that causes the exception.",

+ allow_types=str)

+ CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_optional("value", "Value that causes the exception.",

+ allow_types=[])

+ CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_with("path", "Path that causes the exception.",)

+ CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_with("type", "Python type related to the exception.")

+

+ def __init__(self, message: str, *args,

+ inner: Exception | list[Exception] | None = None,

+ internal: bool = False,

+ name: str | None = None,

+ value: Any | None = None,

+ path: str | None = None,

+ type_: type | None = None,

+ init_attrs: dict[str, Any] | None = None,

+ attrs: dict[str, Any] | None = None, **kwargs) -> None:

+ super().__init__(message, *args)

+

+ self._attrs = {

+ CruException.MESSAGE_KEY: message,

+ CruException.INTERNAL_KEY: internal,

+ CruException.INNER_KEY: inner,

+ CruException.NAME_KEY: name,

+ CruException.VALUE_KEY: value,

+ CruException.PATH_KEY: path,

+ CruException.TYPE_KEY: type_,

+ }

+ if init_attrs is not None:

+ self._attrs.update(init_attrs)

+ if attrs is not None:

+ self._attrs.update(attrs)

+ self._attrs.update(kwargs)

+

+ @property

+ def message(self) -> str:

+ return self[CruException.MESSAGE_KEY]

+

+ @property

+ def internal(self) -> bool:

+ return self[CruException.INTERNAL_KEY]

+

+ @property

+ def inner(self) -> list[Exception]:

+ return self[CruException.INNER_KEY]

+

+ @property

+ def name(self) -> str | None:

+ return self[CruException.NAME_KEY]

+

+ @property

+ def value(self) -> Any | None:

+ return self[CruException.VALUE_KEY]

+

+ @property

+ def path(self) -> str | None:

+ return self[CruException.PATH_KEY]

+

+ @property

+ def type(self) -> type | None:

+ return self[CruException.TYPE_KEY]

+

+ def _get_attr_list_recursive(self, name: str, depth: int, max_depth: int, l: list[Any]):

+ if 0 < max_depth < depth + 1:

+ return

+ a = self._attrs.get(name, None)

+ if a is not None:

+ l.append(a)

+ for i in self.inner:

+ if isinstance(i, CruException):

+ i._get_attr_list_recursive(name, depth + 1, max_depth, l)

+

+ def get_attr_list_recursive(self, name: str, /, max_depth: int = -1) -> list[Any]:

+ l = []

+ self._get_attr_list_recursive(name, 0, max_depth, l)

+ return l

+

+ def get_optional_attr(self, name: str, max_depth: int = -1) -> Any | None:

+ l = self.get_attr_list_recursive(name, max_depth)

+ return l[0] if len(l) > 0 else None

+

+ def __getitem__(self, name: str) -> Any | None:

+ return self.get_optional_attr(name)

+

+

+class CruInternalLogicError(CruException):

+ def __init__(self, message: str, *args, **kwargs) -> None:

+ super().__init__(message, *args, internal=True, **kwargs)

+

+

+class UserFriendlyException(CruException):

+ USER_MESSAGE_KEY = "user_message"

+

+ CRU_EXCEPTION_ATTR_DEF_REGISTRY.register(

+ CruExceptionAttrDef(USER_MESSAGE_KEY, "Message describing the exception, but with user-friendly language."))

+

+ def __init__(self, message: str, user_message: str | None = None, *args, **kwargs) -> None:

+ if user_message is None:

+ user_message = message

+ super().__init__(message, *args, init_attrs={UserFriendlyException.USER_MESSAGE_KEY: user_message}, **kwargs)

+

+ @property

+ def user_message(self) -> str:

+ return self[UserFriendlyException.USER_MESSAGE_KEY]

+from abc import ABCMeta, abstractmethod

+from typing import TypeVar, Generic, NoReturn, Callable

+

+from cru.excp import CruException, CRU_EXCEPTION_ATTR_DEF_REGISTRY

+

+R = TypeVar("R")

+

+

+class ParseException(CruException):

+ LINE_NUMBER_KEY = "line_number"

+

+ CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_with(LINE_NUMBER_KEY, "Line number of the error.")

+

+

+class Parser(Generic[R], metaclass=ABCMeta):

+ def __init__(self, name: str) -> None:

+ self._name = name

+

+ @property

+ def name(self) -> str:

+ return self._name

+

+ @abstractmethod

+ def parse(self, s: str) -> R:

+ raise NotImplementedError()

+

+ def raise_parse_exception(self, s: str, line_number: int | None = None) -> NoReturn:

+ a = f" at line {line_number}" if line_number is not None else ""

+ raise ParseException(f"Parser {self.name} failed{a}, {s}")

+

+

+class SimpleLineConfigParser(Parser[list[tuple[str, str]]]):

+ def __init__(self) -> None:

+ super().__init__(type(self).__name__)

+

+ def _parse(self, s: str, f: Callable[[str, str], None]) -> None:

+ for ln, line in enumerate(s.splitlines()):

+ line_number = ln + 1

+ # check if it's a comment

+ if line.strip().startswith("#"):

+ continue

+ # check if there is a '='

+ if line.find("=") == -1:

+ self.raise_parse_exception(f"There is even no '='!", line_number)

+ # split at first '='

+ key, value = line.split("=", 1)

+ key = key.strip()

+ value = value.strip()

+ f(key, value)

+

+ def parse(self, s: str) -> list[tuple[str, str]]:

+ items = []

+ self._parse(s, lambda key, value: items.append((key, value)))

+ return items

+

+ def parse_to_dict(self, s: str, /, allow_override: bool = False) -> tuple[dict[str, str], list[tuple[str, str]]]:

+ d = {}

+ duplicate = []

+

+ def add(key: str, value: str) -> None:

+ if key in d:

+ if allow_override:

+ duplicate.append((key, d[key]))

+ d[key] = value

+ else:

+ self.raise_parse_exception(f"Key '{key}' already exists!", None)

+ d[key] = value

+

+ self._parse(s, add)

+ return d, duplicate

+from pathlib import Path

+import os

+

+from .excp import CruException

+

+

+class ApplicationPathError(CruException):

+ def __init__(self, message: str, p: str | Path, *args, **kwargs):

+ super().__init__(message, *args, path=str(p), **kwargs)

+

+

+def check_parents_dir(p: str | Path, /, must_exist: bool = False) -> bool:

+ p = Path(p) if isinstance(p, str) else p

+ for p in reversed(p.parents):

+ if not p.exists() and not must_exist:

+ return False

+ if not p.is_dir():

+ raise ApplicationPathError("Parents path should be a dir.", p)

+ return True

+

+

+class ApplicationPath:

+ def __init__(self, p: str | Path, is_dir: bool) -> None:

+ self._path = Path(p) if isinstance(p, str) else p

+ self._is_dir = is_dir

+

+ @property

+ def path(self) -> Path:

+ return self._path

+

+ @property

+ def is_dir(self) -> bool:

+ return self._is_dir

+

+ def check_parents(self, must_exist: bool = False) -> bool:

+ return check_parents_dir(self._path.parent, must_exist)

+

+ def check_self(self, must_exist: bool = False) -> bool:

+ if not self.check_parents(must_exist):

+ return False

+ if not self.path.exists():

+ if not must_exist: return False

+ raise ApplicationPathError("Mot exist.", self.path)

+ if self.is_dir:

+ if not self.path.is_dir():

+ raise ApplicationPathError("Should be a directory, but not.", self.path)

+ else:

+ return False

+ else:

+ if not self.path.is_file():

+ raise ApplicationPathError("Should be a file, but not.", self.path)

+ else:

+ return False

+

+ def ensure(self, create_file: bool = False) -> None:

+ e = self.check_self(False)

+ if not e:

+ os.makedirs(self.path.parent, exist_ok=True)

+ if self.is_dir:

+ os.mkdir(self.path)

+ elif create_file:

+ with open(self.path, "w") as f:

+ f.write("")

+import shutil

+

+

+class DockerController:

+ DOCKER_BIN_NAME = "docker"

+

+ def __init__(self, docker_bin: None | str = None) -> None:

+ self._docker_bin = docker_bin

+

+ @property

+ def docker_bin(self) -> str:

+ if self._docker_bin is None:

+ self._docker_bin = shutil.which(self.DOCKER_BIN_NAME)

+ return self._docker_bin

+

+from typing import Literal, Any, cast, ClassVar

+import os

+from os.path import join, basename, dirname

+import subprocess

+import re

+import json

+import jsonschema

+

+from crupest.template2 import Template2

+from crupest.tui import Paths, UserFriendlyException, create_dir_if_not_exists, console, Confirm, ensure_dir

+from crupest.ui_base import file_name_style

+

+

+def restart_nginx(force=False) -> bool:

+ if not force:

+ p = subprocess.run(['docker', "container", "ls",

+ "-f", "name=nginx", "-q"], capture_output=True)

+ container: str = p.stdout.decode("utf-8")

+ if len(container.strip()) == 0:

+ return False

+ subprocess.run(['docker', 'restart', 'nginx'])

+ return True

+

+

+_server_schema_filename = "server.schema.json"

+

+with open(join(Paths.nginx2_template_dir, _server_schema_filename)) as f:

+ server_json_schema = json.load(f)

+

+

+_domain_template_filename = "domain.conf.template"

+

+NginxSourceFileType = Literal["global", "domain", "http", "https"]

+

+

+class NginxSourceFile:

+ def __init__(self, path: str) -> None:

+ """

+ path: relative to nginx2_template_dir

+ """

+ self._path = path

+ is_template = path.endswith(".template")

+ self._is_template = is_template

+ filename = basename(path)

+ self.name = filename[:-len(".template")] if is_template else filename

+ if is_template:

+ self._template = Template2.from_file(

+ join(Paths.nginx2_template_dir, path))

+ else:

+ with open(join(Paths.nginx2_template_dir, path)) as f:

+ self._content = f.read()

+

+ self._scope: NginxSourceFileType = self._calc_scope()

+

+ @property

+ def is_template(self) -> bool:

+ return self._is_template

+

+ @property

+ def content(self) -> str:

+ if self._is_template:

+ raise Exception(f"{self._path} is a template file")

+ return self._content

+

+ @property

+ def template(self) -> Template2:

+ if not self._is_template:

+ raise Exception(f"{self._path} is not a template file")

+ return cast(Template2, self._template)

+

+ @property

+ def global_target_filename(self) -> str:

+ if self.scope != "global":

+ raise Exception(f"{self._path} is not a global file")

+ if self.is_template:

+ return basename(self._path)[:-len(".template")]

+ else:

+ return basename(self._path)

+

+ def _calc_scope(self) -> NginxSourceFileType:

+ f = basename(self._path)

+ d = basename(dirname(self._path))

+ if f == _domain_template_filename:

+ return "domain"

+ elif d in ["global", "http", "https"]:

+ return cast(Literal["global", "http", "https"], d)

+ else:

+ raise Exception(f"Unknown scope for {self._path}")

+

+ @property

+ def scope(self) -> NginxSourceFileType:

+ return self._scope

+

+

+_domain_template_source = NginxSourceFile(_domain_template_filename)

+

+_client_max_body_size_source = NginxSourceFile(

+ "global/client-max-body-size.conf")

+_forbid_unknown_domain_source = NginxSourceFile(

+ "global/forbid-unknown-domain.conf")

+_ssl_template_source = NginxSourceFile("global/ssl.conf.template")

+_websocket_source = NginxSourceFile("global/websocket.conf")

+

+_http_444_source = NginxSourceFile("http/444.segment")

+_http_redirect_to_https_source = NginxSourceFile(

+ "http/redirect-to-https.segment")

+

+_https_redirect_template_source = NginxSourceFile(

+ "https/redirect.segment.template")

+_https_reverse_proxy_template_source = NginxSourceFile(

+ "https/reverse-proxy.segment.template")

+_https_static_file_template_source = NginxSourceFile(

+ "https/static-file.segment.template")

+_https_static_file_no_strip_prefix_template_source = NginxSourceFile(

+ "https/static-file.no-strip-prefix.segment.template")

+

+

+class NginxService:

+ def __init__(self, type: str, path: str) -> None:

+ self.type = type

+ self.path = path

+ self._check_path(path)

+

+ @staticmethod

+ def _check_path(path: str) -> None:

+ assert isinstance(path, str)

+ if path == "" or path == "/":

+ return

+ if not path.startswith("/"):

+ raise UserFriendlyException("Service path should start with '/'.")

+ if path.endswith("/"):

+ raise UserFriendlyException(

+ "Service path should not end with '/'.")

+

+ def generate_https_segment(self) -> str:

+ raise NotImplementedError()

+

+

+class NginxRedirectService(NginxService):

+ def __init__(self, path: str, redirect_url: str, redirect_code: int = 307) -> None:

+ if redirect_url.endswith("/"):

+ raise UserFriendlyException(

+ "Redirect URL should not end with '/'.")

+

+ super().__init__("redirect", path)

+

+ self.redirect_url = redirect_url

+ self.redirect_code = redirect_code

+

+ def generate_https_segment(self) -> str:

+ vars = {

+ "PATH": self.path,

+ "REDIRECT_CODE": self.redirect_code,

+ "REDIRECT_URL": self.redirect_url

+ }

+ return _https_redirect_template_source.template.render(vars)

+

+ @staticmethod

+ def from_json(json: dict[str, Any]) -> "NginxRedirectService":

+ path = json["path"]

+ redirect_url = json["to"]

+ redirect_code = json.get("code", 307)

+ assert isinstance(path, str)

+ assert isinstance(redirect_url, str)

+ assert isinstance(redirect_code, int)

+ return NginxRedirectService(path, redirect_url, redirect_code)

+

+

+class NginxReverseProxyService(NginxService):

+

+ _upstream_regex: ClassVar[re.Pattern[str]] = re.compile(

+ r"^[-_0-9a-zA-Z]+:[0-9]+$")

+

+ def __init__(self, path: str, upstream: str) -> None:

+ if not self._upstream_regex.match(upstream):

+ raise UserFriendlyException(

+ f"Invalid upstream format: {upstream}.")

+

+ super().__init__("reverse-proxy", path)

+

+ self.upstream = upstream

+

+ def generate_https_segment(self) -> str:

+ vars = {

+ "PATH": self.path,

+ "UPSTREAM": self.upstream

+ }

+ return _https_reverse_proxy_template_source.template.render(vars)

+

+ @staticmethod

+ def from_json(json: dict[str, Any]) -> "NginxReverseProxyService":

+ path = json["path"]

+ upstream = json["upstream"]

+ assert isinstance(path, str)

+ assert isinstance(upstream, str)

+ return NginxReverseProxyService(path, upstream)

+

+

+class NginxStaticFileService(NginxService):

+ def __init__(self, path: str, root: str, no_strip_prefix: bool = False) -> None:

+ super().__init__("static-file", path)

+

+ self.root = root

+ self.no_strip_prefix = no_strip_prefix

+

+ def generate_https_segment(self) -> str:

+ vars = {

+ "PATH": self.path,

+ "ROOT": self.root,

+ }

+ if self.no_strip_prefix:

+ return _https_static_file_no_strip_prefix_template_source.template.render(vars)

+ else:

+ return _https_static_file_template_source.template.render(vars)

+

+ @staticmethod

+ def from_json(json: dict[str, Any]) -> "NginxStaticFileService":

+ path = json["path"]

+ root = json["root"]

+ no_strip_prefix = json.get("no_strip_prefix", False)

+ assert isinstance(path, str)

+ assert isinstance(root, str)

+ assert isinstance(no_strip_prefix, bool)

+ return NginxStaticFileService(path, root, no_strip_prefix)

+

+

+def nginx_service_from_json(json: dict[str, Any]) -> NginxService:

+ type = json["type"]

+ if type == "redirect":

+ return NginxRedirectService.from_json(json)

+ elif type == "reverse-proxy":

+ return NginxReverseProxyService.from_json(json)

+ elif type == "static-file":

+ return NginxStaticFileService.from_json(json)

+ else:

+ raise UserFriendlyException(f"Invalid crupest type: {type}.")

+

+

+def _prepend_indent(text: str, indent: str = " " * 4) -> str:

+ lines = text.split("\n")

+ for i in range(len(lines)):

+ if lines[i] != "":

+ lines[i] = indent + lines[i]

+ return "\n".join(lines)

+

+

+class NginxDomain:

+ def __init__(self, domain: str, services: list[NginxService] = []) -> None:

+ self.domain = domain

+ self.services = services

+

+ def add_service(self, service: NginxService) -> None:

+ self.services.append(service)

+

+ def generate_http_segment(self) -> str:

+ if len(self.services) == 0:

+ return _http_444_source.content

+ else:

+ return _http_redirect_to_https_source.content

+

+ def generate_https_segment(self) -> str:

+ return "\n\n".join([s.generate_https_segment() for s in self.services])

+

+ def generate_config(self) -> str:

+ vars = {

+ "DOMAIN": self.domain,

+ "HTTP_SEGMENT": _prepend_indent(self.generate_http_segment()),

+ "HTTPS_SEGMENT": _prepend_indent(self.generate_https_segment()),

+ }

+ return _domain_template_source.template.render(vars)

+

+ def generate_config_file(self, path: str) -> None:

+ with open(path, "w") as f:

+ f.write(self.generate_config())

+

+ @staticmethod

+ def from_json(root_domain: str, json: dict[str, Any]) -> "NginxDomain":

+ name = json["name"]

+ assert isinstance(name, str)

+ if name == "@" or name == "":

+ domain = root_domain

+ else:

+ domain = f"{name}.{root_domain}"

+ assert isinstance(json["services"], list)

+ services = [nginx_service_from_json(s) for s in json["services"]]

+ return NginxDomain(domain, services)

+

+

+def check_nginx_config_schema(json: Any) -> None:

+ jsonschema.validate(json, server_json_schema)

+

+

+class NginxServer:

+ def __init__(self, root_domain: str) -> None:

+ self.root_domain = root_domain

+ self.domains: list[NginxDomain] = []

+

+ def add_sub_domain(self, sub_domain: str, services: list[NginxService]) -> None:

+ if sub_domain == "" or sub_domain == "@":

+ domain = self.root_domain

+ else:

+ domain = f"{sub_domain}.{self.root_domain}"

+ self.domains.append(NginxDomain(domain, services))

+

+ def generate_ssl(self) -> str:

+ return _ssl_template_source.template.render({

+ "ROOT_DOMAIN": self.root_domain

+ })

+

+ def generate_global_files(self, d: str) -> None:

+ for source in [_client_max_body_size_source, _forbid_unknown_domain_source, _websocket_source]:

+ with open(join(d, source.name), "w") as f:

+ f.write(source.content)

+ with open(join(d, _ssl_template_source.name), "w") as f:

+ f.write(self.generate_ssl())

+

+ def generate_domain_files(self, d: str) -> None:

+ for domain in self.domains:

+ domain.generate_config_file(join(d, f"{domain.domain}.conf"))

+

+ def generate_config(self, d: str) -> None:

+ create_dir_if_not_exists(d)

+ self.generate_global_files(d)

+

+ def get_allowed_files(self) -> list[str]:

+ files = []

+ for source in [_client_max_body_size_source, _forbid_unknown_domain_source, _ssl_template_source, _websocket_source]:

+ files.append(source.name)

+ for domain in self.domains:

+ files.append(f"{domain.domain}.conf")

+ return files

+

+ def check_bad_files(self, d: str) -> list[str]:

+ allowed_files = self.get_allowed_files()

+ bad_files = []

+ if not ensure_dir(d, must_exist=False):

+ return []

+ for path in os.listdir(d):

+ if path not in allowed_files:

+ bad_files.append(path)

+ return bad_files

+

+ @staticmethod

+ def from_json(root_domain: str, json: dict[str, Any]) -> "NginxServer":

+ check_nginx_config_schema(json)

+ server = NginxServer(root_domain)

+ sub_domains = json["domains"]

+ assert isinstance(sub_domains, list)

+ server.domains = [NginxDomain.from_json(

+ root_domain, d) for d in sub_domains]

+ return server

+

+ @staticmethod

+ def from_json_str(root_domain: str, json_str: str) -> "NginxServer":

+ return NginxServer.from_json(root_domain, json.loads(json_str))

+

+ def go(self):

+ bad_files = self.check_bad_files(Paths.nginx_generated_dir)

+ if len(bad_files) > 0:

+ console.print(

+ "WARNING: It seems there are some bad conf files in the nginx config directory:", style="yellow")

+ for bad_file in bad_files:

+ console.print(bad_file, style=file_name_style)

+ to_delete = Confirm.ask(

+ "They will affect nginx in a [red]bad[/] way. Do you want to delete them?", default=True, console=console)

+ if to_delete:

+ for file in bad_files:

+ os.remove(join(Paths.nginx_generated_dir, file))

+ create_dir_if_not_exists(Paths.generated_dir)

+ if not ensure_dir(Paths.nginx_generated_dir, must_exist=False):

+ os.mkdir(Paths.nginx_generated_dir)

+ console.print(

+ f"Nginx config directory created at [magenta]{Paths.nginx_generated_dir}[/]", style="green")

+ self.generate_config(Paths.nginx_generated_dir)

+ console.print("Nginx config generated.", style="green")

+ if restart_nginx():

+ console.print('Nginx restarted.', style="green")

+import re

+import os.path

+

+

+def check_debian_derivative_version(name: str) -> None | str:

+ if not os.path.isfile("/etc/os-release"):

+ return None

+ with open("/etc/os-release", "r") as f:

+ content = f.read()

+ if not f"ID={name}" in content:

+ return None

+ m = re.search(r'VERSION_ID="(.+)"', content)

+ if m is None: return None

+ return m.group(1)

+

+

+def check_ubuntu_version() -> None | str:

+ return check_debian_derivative_version("ubuntu")

+

+

+def check_debian_version() -> None | str:

+ return check_debian_derivative_version("debian")

+import random

+import secrets

+import string

+import uuid

+from abc import abstractmethod, ABCMeta

+from collections.abc import Mapping, Callable

+from typing import Any, ClassVar, Literal, TypeVar, Generic, ParamSpec

+

+from .excp import CruInternalLogicError, CruException, CRU_EXCEPTION_ATTR_DEF_REGISTRY

+

+

+def _str_case_in(s: str, case: bool, l: list[str]) -> bool:

+ if case:

+ return s in l

+ else:

+ return s.lower() in [s.lower() for s in l]

+

+

+_ValueTypeForward = type["ValueType"]

+

+T = TypeVar("T")

+

+

+class _ValueErrorMixin:

+ VALUE_TYPE_KEY = "value_type"

+

+ CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_with(

+ VALUE_TYPE_KEY,

+ "The type of the value that causes the exception."

+ )

+

+

+class ValidationError(CruException, _ValueErrorMixin):

+ def __init__(self, message: str, value: Any, value_type: _ValueTypeForward[T] | None, *args, **kwargs):

+ super().__init__(message, *args, value=value, type_=value_type.type, init_attrs={

+ ValidationError.VALUE_TYPE_KEY: value_type,

+ }, **kwargs)

+

+ @property

+ def value_type(self) -> _ValueTypeForward[T] | None:

+ return self[ValidationError.VALUE_TYPE_KEY]

+

+

+class ValueStringConvertionError(CruException, _ValueErrorMixin):

+ def __init__(self, message: str, value: Any, value_type: _ValueTypeForward[T] | None, *args,

+ **kwargs):

+ super().__init__(message, *args, value=value, type_=value_type.type, init_attrs={

+ ValueStringConvertionError.VALUE_TYPE_KEY: value_type,

+ }, **kwargs)

+

+ @property

+ def value_type(self) -> _ValueTypeForward[T] | None:

+ return self[ValueStringConvertionError.VALUE_TYPE_KEY]

+

+

+class ValueType(Generic[T], metaclass=ABCMeta):

+ @staticmethod

+ def case_sensitive_to_str(case_sensitive: bool) -> str:

+ return f"case-{'' if case_sensitive else 'in'}sensitive"

+

+ def __init__(self, name: str) -> None:

+ self._name = name

+ self._type = type("T")

+

+ @property

+ def name(self) -> str:

+ return self._name

+

+ @property

+ def type(self) -> type:

+ return self._type

+

+ def is_instance_of_value_type(self, value: Any) -> bool:

+ return isinstance(value, self.type)

+

+ def _do_check_value(self, value: Any) -> tuple[True, T] | tuple[False, None | str]:

+ return True, value

+

+ def check_value(self, value: Any) -> T:

+ if not isinstance(value, self.type):

+ raise ValidationError("Value type is wrong.", value, self)

+ ok, v_or_err = self._do_check_value(value)

+ if ok:

+ return v_or_err

+ else:

+ raise ValidationError(v_or_err or "Value is not valid.", value, self)

+

+ @abstractmethod

+ def _do_check_str_format(self, s: str) -> bool | tuple[bool, str]:

+ """

+ Return None for no error. Otherwise, return error message.

+ """

+ raise NotImplementedError()

+

+ def check_str_format(self, s: str) -> None:

+ ok, err = self._do_check_str_format(s)

+ if ok is None: raise CruInternalLogicError("_do_check_str_format should not return None.")

+ if ok: return

+ if err is None:

+ err = "Invalid value str format."

+ raise ValueStringConvertionError(err, s, value_type=self)

+

+ @abstractmethod

+ def _do_convert_value_to_str(self, value: T) -> str:

+ raise NotImplementedError()

+

+ def convert_value_to_str(self, value: T) -> str:

+ self.check_value(value)

+ return self._do_convert_value_to_str(value)

+

+ @abstractmethod

+ def _do_convert_str_to_value(self, s: str) -> T:

+ raise NotImplementedError()

+

+ def convert_str_to_value(self, s: str) -> T:

+ self.check_str_format(s)

+ return self._do_convert_str_to_value(s)

+

+ def check_value_or_try_convert_from_str(self, value_or_str: Any) -> T:

+ try:

+ return self.check_value(value_or_str)

+ except ValidationError as e:

+ if isinstance(value_or_str, str):

+ return self.convert_str_to_value(value_or_str)

+ else:

+ raise ValidationError("Value is not valid and is not a str.", value_or_str, self,

+ inner=e)

+

+

+class TextValueType(ValueType[str]):

+ def __init__(self) -> None:

+ super().__init__("text")

+

+ def _do_check_str_format(self, s):

+ return True

+

+ def _do_convert_value_to_str(self, value):

+ return value

+

+ def _do_convert_str_to_value(self, s):

+ return s

+

+

+class IntegerValueType(ValueType[int]):

+

+ def __init__(self) -> None:

+ super().__init__("integer")

+

+ def _do_check_str_format(self, s):

+ try:

+ int(s)

+ return True

+ except ValueError:

+ return False

+

+ def _do_convert_value_to_str(self, value):

+ return str(value)

+

+ def _do_convert_str_to_value(self, s):

+ return int(s)

+

+

+class FloatValueType(ValueType[float]):

+ def __init__(self) -> None:

+ super().__init__("float")

+

+ def _do_check_str_format(self, s):

+ try:

+ float(s)

+ return True

+ except ValueError:

+ return False

+

+ def _do_convert_value_to_str(self, value):

+ return str(value)

+

+ def _do_convert_str_to_value(self, s):

+ return float(s)

+

+

+class BooleanValueType(ValueType[bool]):

+ DEFAULT_TRUE_LIST: ClassVar[list[str]] = ["true", "yes", "y", "on", "1"]

+ DEFAULT_FALSE_LIST: ClassVar[list[str]] = ["false", "no", "n", "off", "0"]

+

+ def __init__(self, *, case_sensitive=False, true_list: None | list[str] = None,

+ false_list: None | list[str] = None) -> None:

+ super().__init__("boolean")

+ self._case_sensitive = case_sensitive

+ self._valid_true_strs: list[str] = true_list or BooleanValueType.DEFAULT_TRUE_LIST

+ self._valid_false_strs: list[str] = false_list or BooleanValueType.DEFAULT_FALSE_LIST

+

+ @property

+ def case_sensitive(self) -> bool:

+ return self._case_sensitive

+

+ @property

+ def valid_true_strs(self) -> list[str]:

+ return self._valid_true_strs

+

+ @property

+ def valid_false_strs(self) -> list[str]:

+ return self._valid_false_strs

+

+ @property

+ def valid_boolean_strs(self) -> list[str]:

+ return self._valid_true_strs + self._valid_false_strs

+

+ def _do_check_str_format(self, s):

+ if _str_case_in(s, self.case_sensitive, self.valid_boolean_strs): return True

+ return False, f"Not a valid boolean string ({ValueType.case_sensitive_to_str(self.case_sensitive)}). Valid string of true: {' '.join(self._valid_true_strs)}. Valid string of false: {' '.join(self._valid_false_strs)}. All is case insensitive."

+

+ def _do_convert_value_to_str(self, value):

+ return "True" if value else "False"

+

+ def _do_convert_str_to_value(self, s):

+ return _str_case_in(s, self.case_sensitive, self._valid_true_strs)

+

+

+class EnumValueType(ValueType[str]):

+ def __init__(self, valid_values: list[str], /, case_sensitive=False) -> None:

+ s = ' | '.join([f'"{v}"' for v in valid_values])

+ self._valid_value_str = f'[ {s} ]'

+ super().__init__(f"enum{self._valid_value_str}")

+ self._case_sensitive = case_sensitive

+ self._valid_values = valid_values

+

+ @property

+ def case_sensitive(self) -> bool:

+ return self._case_sensitive

+

+ @property

+ def valid_values(self) -> list[str]:

+ return self._valid_values

+

+ def _do_check_value(self, value):

+ ok, err = self._do_check_str_format(value)

+ return ok, (value if ok else err)

+

+ def _do_check_str_format(self, s):

+ if _str_case_in(s, self.case_sensitive, self.valid_values): return True

+ return False, f"Value is not in valid values ({ValueType.case_sensitive_to_str(self.case_sensitive)}): {self._valid_value_str}"

+

+ def _do_convert_value_to_str(self, value):

+ return value

+

+ def _do_convert_str_to_value(self, s):

+ return s

+

+

+TEXT_VALUE_TYPE = TextValueType()

+INTEGER_VALUE_TYPE = IntegerValueType()

+BOOLEAN_VALUE_TYPE = BooleanValueType()

+

+P = ParamSpec('P')

+

+

+class ValueGenerator(Generic[T, P]):

+ INTERACTIVE_KEY: ClassVar[Literal["interactive"]] = "interactive"

+

+ def __init__(self, f: Callable[P, T], /, attributes: None | Mapping[str, Any] = None) -> None:

+ self._f = f

+ self._attributes = attributes or {}

+

+ @property

+ def f(self) -> Callable[P, T]:

+ return self._f

+

+ @property

+ def attributes(self) -> Mapping[str, Any]:

+ return self._attributes

+

+ def generate(self, *args, **kwargs) -> T:

+ return self._f(*args, **kwargs)

+

+ def __call__(self, *args, **kwargs):

+ return self._f(*args, **kwargs)

+

+ @property

+ def interactive(self) -> bool:

+ return self._attributes.get(ValueGenerator.INTERACTIVE_KEY, False)

+

+ @staticmethod

+ def create_interactive(f: Callable[P, T], interactive: bool = True, /,

+ attributes: None | Mapping[str, Any] = None) -> "ValueGenerator[T, P]":

+ return ValueGenerator(f, dict({ValueGenerator.INTERACTIVE_KEY: interactive}, **(attributes or {})))

+

+

+class UuidValueGenerator(ValueGenerator[str, []]):

+ def __init__(self) -> None:

+ super().__init__(lambda: str(uuid.uuid4()))

+

+

+class RandomStringValueGenerator(ValueGenerator[str, []]):

+ @staticmethod

+ def _create_generate_ramdom_func(length: int, secure: bool) -> Callable[str, []]:

+ random_choice = secrets.choice if secure else random.choice

+

+ def generate_random_string():

+ characters = string.ascii_letters + string.digits

+ random_string = ''.join(random_choice(characters) for _ in range(length))

+ return random_string

+

+ return generate_random_string

+

+ def __init__(self, length: int, secure: bool) -> None:

+ super().__init__(RandomStringValueGenerator._create_generate_ramdom_func(length, secure))

+

+

+UUID_VALUE_GENERATOR = UuidValueGenerator()

+from .path import *

+from rich.prompt import Prompt, Confirm

+from urllib.request import urlretrieve

+import subprocess

+from datetime import datetime

+

+

+def backup_restore(http_url_or_path, /, console):

+ url = http_url_or_path

+ if len(url) == 0:

+ raise Exception("You specify an empty url. Abort.")

+ if url.startswith("http://") or url.startswith("https://"):

+ download_path = os.path.join(tmp_dir, "data.tar.xz")

+ if os.path.exists(download_path):

+ to_remove = Confirm.ask(

+ f"I want to download to [cyan]{download_path}[/]. However, there is a file already there. Do you want to remove it first", default=False, console=console)

+ if to_remove:

+ os.remove(download_path)

+ else:

+ raise Exception(

+ "Aborted! Please check the file and try again.")

+ urlretrieve(url, download_path)

+ url = download_path

+ subprocess.run(["sudo", "tar", "-xJf", url, "-C", project_dir], check=True)

+

+

+def backup_backup(path, /, console):

+ ensure_backup_dir()

+ now = datetime.utcnow().isoformat(timespec="seconds") + "Z"

+ if path is None:

+ path = Prompt.ask(

+ "You don't specify the path to backup to. Please specify one. http and https are NOT supported", console=console, default=os.path.join(backup_dir, now + ".tar.xz"))

+ if len(path) == 0:

+ raise Exception("You specify an empty path. Abort!")

+ if os.path.exists(path):

+ raise Exception(

+ "A file is already there. Please remove it first. Abort!")

+ subprocess.run(

+ ["sudo", "tar", "-cJf", path, "data", "-C", project_dir],

+ check=True

+ )

+from typing import Literal, cast

+import os

+from os.path import join

+import subprocess

+from cryptography.x509 import load_pem_x509_certificate, DNSName, SubjectAlternativeName

+from cryptography.x509.oid import ExtensionOID

+from .tui import Paths, ensure_file, create_dir_if_not_exists, console

+

+CertbotAction = Literal['create', 'expand', 'shrink', 'renew']

+

+

+class Certbot:

+ def __init__(self, root_domain: str, subdomains: list[str]) -> None:

+ """

+ subdomain: like ["a", "b.c", ...]

+ """

+ self.root_domain = root_domain

+ self.subdomains = subdomains

+ self.domains = [

+ root_domain, *[f"{subdomain}.{root_domain}" for subdomain in subdomains]]

+

+ def generate_command(self, action: CertbotAction, /, test=False, no_docker=False, *, standalone=None, email=None, agree_tos=False) -> str:

+ add_domain_option = True

+ if action == 'create':

+ if standalone == None:

+ standalone = True

+ certbot_action = "certonly"

+ elif action == 'expand' or action == 'shrink':

+ if standalone == None:

+ standalone = False

+ certbot_action = "certonly"

+ elif action == 'renew':

+ if standalone == None:

+ standalone = False

+ add_domain_option = False

+ certbot_action = "renew"

+ else:

+ raise ValueError('Invalid action')

+

+ if no_docker:

+ command = "certbot "

+ else:

+ expose_segment = ' -p "0.0.0.0:80:80"'

+ web_root_segment = f' -v "{Paths.project_abs_path}/data/certbot/webroot:/var/www/certbot"'

+ command = f'docker run -it --rm --name certbot -v "{Paths.project_abs_path}/data/certbot/certs:/etc/letsencrypt" -v "{Paths.project_abs_path}/data/certbot/data:/var/lib/letsencrypt"{ expose_segment if standalone else web_root_segment} certbot/certbot '

+

+ command += certbot_action

+

+ if standalone:

+ command += " --standalone"

+ else:

+ command += ' --webroot -w /var/www/certbot'

+

+ if add_domain_option:

+ command += f' -d {" -d ".join(self.domains)}'

+

+ if email is not None:

+ command += f' --email {email}'

+

+ if agree_tos:

+ command += ' --agree-tos'

+

+ if test:

+ command += " --test-cert --dry-run"

+

+ return command

+

+ def get_cert_path(self) -> str:

+ return join(Paths.data_dir, "certbot", "certs", "live", self.root_domain, "fullchain.pem")

+

+ def get_cert_actual_domains(self, cert_path: str | None = None) -> None | list[str]:

+ if cert_path is None:

+ cert_path = self.get_cert_path()

+

+ if not ensure_file(cert_path):

+ return None

+

+ with open(cert_path, 'rb') as f:

+ cert = load_pem_x509_certificate(f.read())

+ ext = cert.extensions.get_extension_for_oid(

+ ExtensionOID.SUBJECT_ALTERNATIVE_NAME)

+ domains: list[str] = cast(

+ SubjectAlternativeName, ext.value).get_values_for_type(DNSName)

+

+ # This weird code is to make sure the root domain is the first one

+ if self.root_domain in domains:

+ domains.remove(self.root_domain)

+ domains = [self.root_domain, *domains]

+

+ return domains

+

+ def print_create_cert_message(self):

+ console.print(

+ "Looks like you haven't run certbot to get the init ssl certificates. You may want to run following code to get one:", style="cyan")

+ console.print(self.generate_command("create"),

+ soft_wrap=True, highlight=False)

+

+ def check_ssl_cert(self, tmp_dir: str = Paths.tmp_dir):

+ cert_path = self.get_cert_path()

+ tmp_cert_path = join(tmp_dir, "fullchain.pem")

+ console.print("Temporarily copy cert to tmp...", style="yellow")

+ create_dir_if_not_exists(tmp_dir)

+ subprocess.run(

+ ["sudo", "cp", cert_path, tmp_cert_path], check=True)

+ subprocess.run(["sudo", "chown", str(

+ os.geteuid()), tmp_cert_path], check=True)

+ cert_domains = self.get_cert_actual_domains(tmp_cert_path)

+ if cert_domains is None:

+ self.print_create_cert_message()

+ else:

+ cert_domain_set = set(cert_domains)

+ domains = set(self.domains)

+ if not cert_domain_set == domains:

+ console.print(

+ "Cert domains are not equal to host domains. Run following command to recreate it with nginx stopped.", style="red")

+ console.print(self.generate_command(

+ "create", standalone=True), soft_wrap=True, highlight=False)

+ console.print("Remove tmp cert...", style="yellow")

+ os.remove(tmp_cert_path)

+import os

+import typing

+import uuid

+import random

+import string

+from dataclasses import dataclass

+

+from rich.prompt import Prompt

+

+from cru.config import Configuration

+from cru.parsing import SimpleLineConfigParser

+from .path import config_file_path

+

+

+@dataclass

+class ConfigurationMigrationInfo:

+ duplicate_item_in_old_config: list[str]

+ item

+

+

+class OldConfiguration:

+ def __init__(self, items: None | dict[str, str] = None) -> None:

+ self._items = items or {}

+

+ @staticmethod

+ def load_from_str(s: str) -> tuple["OldConfiguration", list[str, str]]:

+ d, duplicate = SimpleLineConfigParser().parse_to_dict(s, True)

+ return OldConfiguration(d), duplicate

+

+ def convert_to_new_config(self) -> Configuration:

+

+

+class ConfigVar:

+ def __init__(self, name: str, description: str, default_value_generator: typing.Callable[[], str] | str, /,

+ default_value_for_ask=str | None):

+ """Create a config var.

+

+ Args:

+ name (str): The name of the config var.

+ description (str): The description of the config var.

+ default_value_generator (typing.Callable[[], str] | str): The default value generator of the config var. If it is a string, it will be used as the input prompt and let user input the value.

+ """

+ self.name = name

+ self.description = description

+ self.default_value_generator = default_value_generator

+ self.default_value_for_ask = default_value_for_ask

+

+ def get_default_value(self, /, console):

+ if isinstance(self.default_value_generator, str):

+ return Prompt.ask(self.default_value_generator, console=console, default=self.default_value_for_ask)

+ else:

+ return self.default_value_generator()

+

+

+config_var_list: list = [

+ ConfigVar("CRUPEST_DOMAIN", "domain name",

+ "Please input your domain name"),

+ ConfigVar("CRUPEST_EMAIL", "admin email address",

+ "Please input your email address"),

+ ConfigVar("CRUPEST_AUTO_BACKUP_COS_SECRET_ID",

+ "access key id for Tencent COS, used for auto backup",

+ "Please input your Tencent COS access key id for backup"),

+ ConfigVar("CRUPEST_AUTO_BACKUP_COS_SECRET_KEY",

+ "access key secret for Tencent COS, used for auto backup",

+ "Please input your Tencent COS access key for backup"),

+ ConfigVar("CRUPEST_AUTO_BACKUP_COS_REGION",

+ "region for Tencent COS, used for auto backup", "Please input your Tencent COS region for backup",

+ "ap-hongkong"),

+ ConfigVar("CRUPEST_AUTO_BACKUP_BUCKET_NAME",

+ "bucket name for Tencent COS, used for auto backup",

+ "Please input your Tencent COS bucket name for backup"),

+ ConfigVar("CRUPEST_GITHUB_USERNAME",

+ "github username for fetching todos", "Please input your github username for fetching todos", "crupest"),

+ ConfigVar("CRUPEST_GITHUB_PROJECT_NUMBER",

+ "github project number for fetching todos", "Please input your github project number for fetching todos",

+ "2"),

+ ConfigVar("CRUPEST_GITHUB_TOKEN",

+ "github token for fetching todos", "Please input your github token for fetching todos"),

+ ConfigVar("CRUPEST_GITHUB_TODO_COUNT",

+ "github todo count", "Please input your github todo count", 10),

+ ConfigVar("CRUPEST_GITHUB_TODO_COUNT",

+ "github todo count", "Please input your github todo count", 10),

+ ConfigVar("CRUPEST_V2RAY_TOKEN",

+ "v2ray user id", generate_uuid),

+ ConfigVar("CRUPEST_V2RAY_PATH",

+ "v2ray path, which will be prefixed by _", generate_uuid),

+ ConfigVar("CRUPEST_FORGEJO_MAILER_USER",

+ "Forgejo SMTP user.", "Please input your Forgejo SMTP user."),

+ ConfigVar("CRUPEST_FORGEJO_MAILER_PASSWD",

+ "Forgejo SMTP password.", "Please input your Forgejo SMTP password."),

+ ConfigVar("CRUPEST_2FAUTH_APP_KEY",

+ "2FAuth App Key.", generate_random_string_32),

+ ConfigVar("CRUPEST_2FAUTH_MAIL_USERNAME",

+ "2FAuth SMTP user.", "Please input your 2FAuth SMTP user."),

+ ConfigVar("CRUPEST_2FAUTH_MAIL_PASSWORD",

+ "2FAuth SMTP password.", "Please input your 2FAuth SMTP password."),

+]

+

+config_var_name_set = set([config_var.name for config_var in config_var_list])

+

+

+def check_config_var_set(needed_config_var_set: set[str]) -> tuple[bool, list[str], list[str]]:

+ more = []

+ less = []

+ for var_name in needed_config_var_set:

+ if var_name not in config_var_name_set:

+ more.append(var_name)

+ for var_name in config_var_name_set:

+ if var_name not in needed_config_var_set:

+ less.append(var_name)

+ return (True if len(more) == 0 else False, more, less)

+

+

+def config_file_exists():

+ return ensure_file(Paths.config_file_path, must_exist=False)

+

+

+def parse_config(str: str) -> dict[str, str]:

+ return ConfigMap().load_from_str(str).to_dict()

+

+

+def get_domain() -> str:

+ if configuration is None:

+ raise ValueError("Config file not found!")

+ return configuration.get_domain()

+

+

+def config_to_str(config: dict) -> str:

+ return "\n".join([f"{key}={value}" for key, value in config.items()])

+

+

+def print_config(console, config: dict) -> None:

+ for key, value in config.items():

+ console.print(f"[magenta]{key}[/] = [cyan]{value}")

+from os.path import *

+from io import StringIO

+import re

+from .nginx import *

+

+

+def generate_dns_zone(domain: str, ip: str, /, ttl: str | int = 600, *, enable_mail: bool = True, dkim: str | None = None) -> str:

+ result = f"$ORIGIN {domain}.\n\n"

+ result += "; A records\n"

+ result += f"@ {ttl} IN A {ip}\n"

+ subdomains = list_subdomain_names()

+ for subdomain in subdomains:

+ result += f"{subdomain} {ttl} IN A {ip}\n"

+

+ if enable_mail:

+ result += "\n; MX records\n"

+ result += f"@ {ttl} IN MX 10 mail.{domain}.\n"

+ result += "\n; SPF record\n"

+ result += f"@ {ttl} IN TXT \"v=spf1 mx ~all\"\n"

+ if dkim is not None:

+ result += "\n; DKIM record\n"

+ result += f"mail._domainkey {ttl} IN TEXT \"{dkim}\""

+ result += "\n; DMARC record\n"

+ result += "_dmarc {ttl} IN TXT \"v=DMARC1; p=none; rua=mailto:dmarc.report@{domain}; ruf=mailto:dmarc.report@{domain}; sp=none; ri=86400\"\n"

+ return result

+

+

+def get_dkim_from_mailserver(domain: str) -> str | None:

+ dkim_path = join(data_dir, "dms/config/opendkim/keys", domain, "mail.txt")

+ if not exists(dkim_path):

+ return None

+

+ p = subprocess.run(["sudo", "cat", dkim_path],

+ capture_output=True, check=True)

+ value = ""

+ for match in re.finditer("\"(.*)\"", p.stdout.decode('utf-8')):

+ value += match.group(1)

+ return value

+

+

+def generate_dns_zone_with_dkim(domain: str, ip: str, /, ttl: str | int = 600) -> str:

+ return generate_dns_zone(domain, ip, ttl, enable_mail=True, dkim=get_dkim_from_mailserver(domain))

+import sys

+from os.path import *

+from urllib.request import *

+from rich.prompt import Confirm

+from .path import *

+from .helper import print_order

+

+

+TOOLS = [("docker-mailserver setup script", "docker-mailserver-setup.sh",

+ "https://raw.githubusercontent.com/docker-mailserver/docker-mailserver/master/setup.sh")]

+

+

+def download_tools(console):

+ # if we are not linux, we prompt the user

+ if sys.platform != "linux":

+ console.print(

+ "You are not running this script on linux. The tools will not work.", style="yellow")

+ if not Confirm.ask("Do you want to continue?", default=False, console=console):

+ return

+

+ for index, script in enumerate(TOOLS):

+ number = index + 1

+ total = len(TOOLS)

+ print_order(number, total, console)

+ name, filename, url = script

+ # if url is callable, call it

+ if callable(url):

+ url = url()

+ path = join(tool_dir, filename)

+ skip = False

+ if exists(path):

+ overwrite = Confirm.ask(

+ f"[cyan]{name}[/] already exists, download and overwrite?", default=False, console=console)

+ if not overwrite:

+ skip = True

+ else:

+ download = Confirm.ask(

+ f"Download [cyan]{name}[/] to [magenta]{path}[/]?", default=True, console=console)

+ if not download:

+ skip = True

+ if not skip:

+ console.print(f"Downloading {name}...")

+ urlretrieve(url, path)

+ os.chmod(path, 0o755)

+ console.print(f"Downloaded {name} to {path}.", style="green")

+ else:

+ console.print(f"Skipped {name}.", style="yellow")

+import os

+import os.path

+from .path import *

+

+

+def run_in_dir(dir: str, func: callable):

+ old_dir = os.path.abspath(os.getcwd())

+ os.chdir(dir)

+ func()

+ os.chdir(old_dir)

+

+

+def run_in_project_dir(func: callable):

+ run_in_dir(project_dir, func)

+

+

+def print_order(number: int, total: int, /, console) -> None:

+ console.print(f"\[{number}/{total}]", end=" ", style="green")

+from os.path import *

+from .path import *

+import urllib

+import subprocess

+

+

+def install_docker():

+ ensure_tmp_dir()

+ get_docker_path = join(tmp_dir, "get-docker.sh")

+ urllib.request.urlretrieve("https://get.docker.com", get_docker_path)

+ os.chmod(get_docker_path, 0o755)

+ subprocess.run(["sudo", "sh", get_docker_path], check=True)

+ subprocess.run(["sudo", "systemctl", "enable",

+ "--now", "docker"], check=True)

+ subprocess.run(["sudo", "usermod", "-aG", "docker",

+ os.getlogin()], check=True)

+from typing import cast

+import json

+import jsonschema

+import os

+from os.path import *

+import shutil

+import subprocess

+from rich.prompt import Confirm

+from cryptography.x509 import *

+from cryptography.x509.oid import ExtensionOID

+from .template import Template

+from .path import *

+

+with open(join(nginx_template_dir, 'server.json')) as f:

+ server = json.load(f)

+

+with open(join(nginx_template_dir, 'server.schema.json')) as f:

+ schema = json.load(f)

+

+jsonschema.validate(server, schema)

+

+non_template_files = ['forbid_unknown_domain.conf', "websocket.conf"]

+

+ssl_template = Template(join(nginx_template_dir, 'ssl.conf.template'))

+root_template = Template(join(

+ nginx_template_dir, 'root.conf.template'))

+static_file_template = Template(join(

+ nginx_template_dir, 'static-file.conf.template'))

+reverse_proxy_template = Template(join(

+ nginx_template_dir, 'reverse-proxy.conf.template'))

+redirect_template = Template(join(

+ nginx_template_dir, 'redirect.conf.template'))

+cert_only_template = Template(join(

+ nginx_template_dir, 'cert-only.conf.template'))

+

+nginx_var_set = set.union(root_template.var_set,

+ static_file_template.var_set, reverse_proxy_template.var_set)

+

+

+def list_subdomain_names() -> list:

+ return [s["subdomain"] for s in server["sites"]]

+

+

+def list_subdomains(domain: str) -> list:

+ return [f"{s['subdomain']}.{domain}" for s in server["sites"]]

+

+

+def list_domains(domain: str) -> list:

+ return [domain, *list_subdomains(domain)]

+

+

+def generate_nginx_config(domain: str, original_config, dest: str) -> None:

+ if not isdir(dest):

+ raise ValueError('dest must be a directory')

+ # copy ssl.conf and https-redirect.conf which need no variable substitution

+ for filename in non_template_files:

+ src = join(nginx_template_dir, filename)

+ dst = join(dest, filename)

+ shutil.copyfile(src, dst)

+ config = {

+ "CRUPEST_DOMAIN": domain,

+ "CRUPEST_V2RAY_TOKEN": original_config["CRUPEST_V2RAY_TOKEN"],

+ "CRUPEST_V2RAY_PATH": original_config["CRUPEST_V2RAY_PATH"]

+ }

+ # generate ssl.conf

+ with open(join(dest, 'ssl.conf'), 'w') as f:

+ f.write(ssl_template.generate(config))

+ # generate root.conf

+ with open(join(dest, f'{domain}.conf'), 'w') as f:

+ root_config = config.copy()

+ root_config["CRUPEST_V2RAY_TOKEN"] = config["CRUPEST_V2RAY_TOKEN"]

+ root_config["CRUPEST_V2RAY_PATH"] = config["CRUPEST_V2RAY_PATH"]

+ f.write(root_template.generate(config))

+ # generate nginx config for each site

+ sites: list = server["sites"]

+ for site in sites:

+ subdomain = site["subdomain"]

+ local_config = config.copy()

+ local_config['CRUPEST_NGINX_SUBDOMAIN'] = subdomain

+ if site["type"] == 'static-file':

+ template = static_file_template

+ local_config['CRUPEST_NGINX_ROOT'] = site["root"]

+ elif site["type"] == 'reverse-proxy':

+ template = reverse_proxy_template

+ local_config['CRUPEST_NGINX_UPSTREAM_SERVER'] = site["upstream"]

+ elif site["type"] == 'redirect':

+ template = redirect_template

+ local_config['CRUPEST_NGINX_URL'] = site["url"]

+ elif site["type"] == 'cert-only':

+ template = cert_only_template

+ else:

+ raise Exception('Invalid site type')

+ with open(join(dest, f'{subdomain}.{domain}.conf'), 'w') as f:

+ f.write(template.generate(local_config))

+

+

+def check_nginx_config_dir(dir_path: str, domain: str) -> list:

+ if not exists(dir_path):

+ return []

+ good_files = [*non_template_files, "ssl.conf", *

+ [f"{full_domain}.conf" for full_domain in list_domains(domain)]]

+ bad_files = []

+ for path in os.listdir(dir_path):

+ file_name = basename(path)

+ if file_name not in good_files:

+ bad_files.append(file_name)

+ return bad_files

+

+

+def restart_nginx(force=False) -> bool:

+ if not force:

+ p = subprocess.run(['docker', "container", "ls",

+ "-f", "name=nginx", "-q"], capture_output=True)

+ container: str = p.stdout.decode("utf-8")

+ if len(container.strip()) == 0:

+ return False

+ subprocess.run(['docker', 'restart', 'nginx'])

+ return True

+

+

+def nginx(domain: str, config, /, console) -> None:

+ bad_files = check_nginx_config_dir(nginx_config_dir, domain)

+ if len(bad_files) > 0:

+ console.print(

+ "WARNING: It seems there are some bad conf files in the nginx config directory:", style="yellow")

+ for bad_file in bad_files:

+ console.print(bad_file, style="cyan")

+ to_delete = Confirm.ask(

+ "They will affect nginx in a [red]bad[/] way. Do you want to delete them?", default=True, console=console)

+ if to_delete:

+ for file in bad_files:

+ os.remove(join(nginx_config_dir, file))

+ console.print(

+ "I have found following var in nginx templates:", style="green")

+ for var in nginx_var_set:

+ console.print(var, style="magenta")

+ if not exists(nginx_config_dir):

+ os.mkdir(nginx_config_dir)

+ console.print(

+ f"Nginx config directory created at [magenta]{nginx_config_dir}[/]", style="green")

+ generate_nginx_config(domain, config, dest=nginx_config_dir)

+ console.print("Nginx config generated.", style="green")

+ if restart_nginx():

+ console.print('Nginx restarted.', style="green")

+

+

+def certbot_command_gen(domain: str, action, /, test=False, no_docker=False, *, standalone=None, email=None, agree_tos=False) -> str:

+ domains = list_domains(domain)

+

+ add_domain_option = True

+ if action == 'create':

+ if standalone == None:

+ standalone = True

+ certbot_action = "certonly"

+ elif action == 'expand':

+ if standalone == None:

+ standalone = False

+ certbot_action = "certonly"

+ elif action == 'renew':

+ if standalone == None:

+ standalone = False

+ add_domain_option = False

+ certbot_action = "renew"

+ else:

+ raise ValueError('Invalid action')

+

+ if no_docker:

+ command = "certbot "

+ else:

+ expose_segment = ' -p "0.0.0.0:80:80"'

+ web_root_segment = ' -v "{project_abs_path}/data/certbot/webroot:/var/www/certbot"'

+ command = f'docker run -it --rm --name certbot -v "{project_abs_path}/data/certbot/certs:/etc/letsencrypt" -v "{project_abs_path}/data/certbot/data:/var/lib/letsencrypt"{ expose_segment if standalone else web_root_segment} certbot/certbot '

+

+ command += certbot_action

+

+ if standalone:

+ command += " --standalone"

+ else:

+ command += ' --webroot -w /var/www/certbot'

+

+ if add_domain_option:

+ command += f' -d {" -d ".join(domains)}'

+

+ if email is not None:

+ command += f' --email {email}'

+

+ if agree_tos:

+ command += ' --agree-tos'

+

+ if test:

+ command += " --test-cert --dry-run"

+

+ return command

+

+

+def get_cert_path(root_domain):

+ return join(data_dir, "certbot", "certs", "live", root_domain, "fullchain.pem")

+

+

+def get_cert_domains(cert_path, root_domain):

+

+ if not exists(cert_path):

+ return None

+

+ if not isfile(cert_path):

+ return None

+

+ with open(cert_path, 'rb') as f:

+ cert = load_pem_x509_certificate(f.read())

+ ext = cert.extensions.get_extension_for_oid(

+ ExtensionOID.SUBJECT_ALTERNATIVE_NAME)

+ domains: list[str] = cast(

+ SubjectAlternativeName, ext.value).get_values_for_type(DNSName)

+ domains.remove(root_domain)

+ domains = [root_domain, *domains]

+ return domains

+

+

+def print_create_cert_message(domain, console):

+ console.print(

+ "Looks like you haven't run certbot to get the init ssl certificates. You may want to run following code to get one:", style="cyan")

+ console.print(certbot_command_gen(domain, "create"),

+ soft_wrap=True, highlight=False)

+

+

+def check_ssl_cert(domain, console):

+ cert_path = get_cert_path(domain)

+ tmp_cert_path = join(tmp_dir, "fullchain.pem")

+ console.print("Temporarily copy cert to tmp...", style="yellow")

+ subprocess.run(

+ ["sudo", "cp", cert_path, tmp_cert_path], check=True)

+ subprocess.run(["sudo", "chown", str(os.geteuid()),

+ tmp_cert_path], check=True)

+ cert_domains = get_cert_domains(tmp_cert_path, domain)

+ if cert_domains is None:

+ print_create_cert_message(domain, console)

+ else:

+ cert_domain_set = set(cert_domains)

+ domains = set(list_domains(domain))

+ if not cert_domain_set == domains:

+ console.print(

+ "Cert domains are not equal to host domains. Run following command to recreate it with nginx stopped.", style="red")

+ console.print(certbot_command_gen(

+ domain, "create", standalone=True), soft_wrap=True, highlight=False)

+ console.print("Remove tmp cert...", style="yellow")

+ os.remove(tmp_cert_path)

+import os

+import os.path

+

+script_dir = os.path.relpath(os.path.dirname(__file__))

+project_dir = os.path.normpath(os.path.join(script_dir, "../../../"))

+project_abs_path = os.path.abspath(project_dir)

+template_dir = os.path.join(project_dir, "template")

+nginx_template_dir = os.path.join(template_dir, "nginx")

+data_dir = os.path.join(project_dir, "data")

+tool_dir = os.path.join(project_dir, "tools")

+tmp_dir = os.path.join(project_dir, "tmp")

+backup_dir = os.path.join(project_dir, "backup")

+config_file_path = os.path.join(data_dir, "config")

+nginx_config_dir = os.path.join(project_dir, "nginx-config")

+log_dir = os.path.join(project_dir, "log")

+

+

+def ensure_file(path: str, /, must_exist: bool = True) -> bool:

+ if must_exist and not os.path.exists(path):

+ raise Exception(f"File {path} does not exist!")

+ if not os.path.exists(path):

+ return False

+ if not os.path.isfile(path):

+ raise Exception(f"{path} is not a file!")

+ return True

+

+

+def ensure_dir(path: str, /, must_exist: bool = True) -> bool:

+ if must_exist and not os.path.exists(path):

+ raise Exception(f"Directory {path} does not exist!")

+ if not os.path.exists(path):

+ return False

+ if not os.path.isdir(path):

+ raise Exception(f"{path} is not a directory!")

+ return True

+

+

+class Paths:

+ script_dir = os.path.relpath(os.path.dirname(__file__))

+ project_dir = os.path.normpath(os.path.join(script_dir, "../../"))

+ project_abs_path = os.path.abspath(project_dir)

+ data_dir = os.path.join(project_dir, "data")

+ config_file_path = os.path.join(data_dir, "config")

+ template_dir = os.path.join(project_dir, "template")

+ tool_dir = os.path.join(project_dir, "tool")

+ tmp_dir = os.path.join(project_dir, "tmp")

+ backup_dir = os.path.join(project_dir, "backup")

+ log_dir = os.path.join(project_dir, "log")

+ template2_dir = os.path.join(project_dir, "template2")

+ nginx2_template_dir = os.path.join(template2_dir, "nginx")

+ generated_dir = os.path.join(project_dir, "generated")

+ nginx_generated_dir = os.path.join(generated_dir, "nginx")

+

+

+def create_dir_if_not_exists(path: str) -> None:

+ if not ensure_dir(path, must_exist=False):

+ os.mkdir(path)

+from os.path import *

+from datetime import datetime

+from rich.prompt import Confirm

+from .path import *

+from .nginx import *

+from .config import *

+from .helper import *

+

+

+def get_template_name_list(console) -> list[str]:

+ console.print("First let's check all the templates...")

+

+ # get all filenames ending with .template

+ template_name_list = [basename(f)[:-len('.template')] for f in os.listdir(

+ template_dir) if f.endswith(".template")]

+ console.print(

+ f"I have found following template files in [magenta]{template_dir}[/]:", style="green")

+ for filename in template_name_list:

+ console.print(f"{filename}.template", style="magenta")

+

+ return template_name_list

+

+

+def data_dir_check(domain, console):

+ if isdir(data_dir):

+ if not exists(join(data_dir, "certbot")):

+ print_create_cert_message(domain, console)

+ else:

+ to_check = Confirm.ask(

+ "I want to check your ssl certs, but I need to sudo. Do you want me check", console=console, default=False)

+ if to_check:

+ check_ssl_cert(domain, console)

+

+

+def template_generate(console):

+ template_name_list = get_template_name_list(console)

+ template_list: list = []

+ config_var_name_set_in_template = set()

+ for template_name in template_name_list:

+ template = Template(join(template_dir, template_name+".template"))

+ template_list.append(template)

+ config_var_name_set_in_template.update(template.var_set)

+

+ console.print(

+ "I have found following variables needed in templates:", style="green")

+ for key in config_var_name_set_in_template:

+ console.print(key, style="magenta")

+

+ # check vars

+ check_success, more, less = check_config_var_set(

+ config_var_name_set_in_template)

+ if len(more) != 0:

+ console.print("There are more variables in templates than in config file:",

+ style="red")

+ for key in more:

+ console.print(key, style="magenta")

+ if len(less) != 0:

+ console.print("Following config vars are not used:",

+ style="yellow")

+ for key in less:

+ console.print(key, style="magenta")

+

+ if not check_success:

+ console.print(

+ "Please check you config vars and make sure the needed ones are defined!", style="red")

+ else:

+ console.print(

+ "Now let's check if they are already generated...")

+

+ conflict = False

+

+ # check if there exists any generated files

+ for filename in template_name_list:

+ if exists(join(project_dir, filename)):

+ console.print(f"Found [magenta]{filename}[/]")

+ conflict = True

+

+ to_gen = True

+ if conflict:

+ to_overwrite = Confirm.ask(

+ "It seems there are some files already generated. Do you want to overwrite them?", console=console, default=False)

+ if not to_overwrite:

+ to_gen = False

+ console.print(

+ "Great! Check the existing files and see you next time!", style="green")

+ else:

+ print("No conflict found. Let's go on!\n")

+

+ if to_gen:

+ console.print("Check for existing config file...")

+

+ # check if there exists a config file

+ if not config_file_exists():

+ config = {}

+ console.print(

+ "No existing config file found. Don't worry. Let's create one!", style="green")

+ for config_var in config_var_list:

+ config[config_var.name] = config_var.get_default_value()

+ config_content = config_to_str(config)

+ # create data dir if not exist

+ if not exists(data_dir):

+ os.mkdir(data_dir)

+ # write config file

+ with open(config_file_path, "w") as f:

+ f.write(config_content)

+ console.print(

+ f"Everything else is auto generated. The config file is written into [magenta]{config_file_path}[/]. You had better keep it safe. And here is the content:", style="green")

+ print_config(console, config)

+ is_ok = Confirm.ask(

+ "If you think it's not ok, you can stop here and edit it. Or let's go on?", console=console, default=True)

+ if not is_ok:

+ console.print(

+ "Great! Check the config file and see you next time!", style="green")

+ to_gen = False

+ else:

+ console.print(

+ "Looks like you have already had a config file. Let's check the content:", style="green")

+ with open(config_file_path, "r") as f:

+ content = f.read()

+ config = parse_config(content)

+ print_config(console, config)

+ missed_config_vars = []

+ for config_var in config_var_list:

+ if config_var.name not in config:

+ missed_config_vars.append(config_var)

+

+ if len(missed_config_vars) > 0:

+ console.print(

+ "Oops! It seems you have missed some keys in your config file. Let's add them!", style="green")

+ for config_var in missed_config_vars:

+ config[config_var.name] = config_var.get_default_value(

+ console)

+ content = config_to_str(config)

+ with open(config_file_path, "w") as f:

+ f.write(content)

+ console.print(

+ f"Here is the new config, it has been written out to [magenta]{config_file_path}[/]:")

+ print_config(console, config)

+ good_enough = Confirm.ask("Is it good enough?",

+ console=console, default=True)

+ if not good_enough:

+ console.print(

+ "Great! Check the config file and see you next time!", style="green")

+ to_gen = False

+

+ domain = get_domain()

+

+ if to_gen:

+ console.print(

+ "Finally, everything is ready. Let's generate the files:", style="green")

+

+ # generate files

+ for index, template in enumerate(template_list):

+ number = index + 1

+ total = len(template_list)

+ print_order(number, total, console)

+ console.print(

+ f"Generating [magenta]{template.template_name}[/]...")

+ content = template.generate(config)

+ with open(join(project_dir, template.template_name), "w") as f:

+ f.write(content)

+

+ # generate nginx config

+ if not exists(nginx_config_dir):

+ to_gen_nginx_conf = Confirm.ask("It seems you haven't generate nginx config. Do you want to generate it?",

+ default=True, console=console)

+ else:

+ # get the latest time of files in nginx template

+ template_time = 0

+ for path in os.listdir(nginx_template_dir):

+ template_time = max(template_time, os.stat(

+ join(nginx_template_dir, path)).st_mtime)

+ console.print(

+ f"Nginx template update time: {datetime.fromtimestamp(template_time)}")

+

+ nginx_config_time = 0

+ for path in os.listdir(nginx_config_dir):

+ nginx_config_time = max(nginx_config_time, os.stat(

+ join(nginx_config_dir, path)).st_mtime)

+ console.print(

+ f"Generated nginx template update time: {datetime.fromtimestamp(nginx_config_time)}")

+ if template_time > nginx_config_time:

+ to_gen_nginx_conf = Confirm.ask("It seems you have updated the nginx template and not regenerate config. Do you want to regenerate the nginx config?",

+ default=True, console=console)

+ else:

+ to_gen_nginx_conf = Confirm.ask("[yellow]It seems you have already generated nginx config. Do you want to overwrite it?[/]",

+ default=False, console=console)

+ if to_gen_nginx_conf:

+ nginx(domain, config, console)

+ data_dir_check(domain, console)

+

+

+def clear(console, /, delete_data_dir=False):

+ template_name_list = get_template_name_list(console)

+ # check root if we have to delete data dir

+ if delete_data_dir and exists(data_dir) and os.geteuid() != 0:

+ console.print(

+ "You need to be root to delete data dir.", style="red")

+ exit(1)

+

+ to_delete = Confirm.ask(

+ "[yellow]Are you sure you want to delete everything? all your data will be lost![/]", default=False, console=console)

+ if to_delete:

+ files_to_delete = []

+ for template_name in template_name_list:

+ f = join(project_dir, template_name)

+ if exists(f):

+ files_to_delete.append(f)

+

+ delete_data_dir = delete_data_dir and exists(

+ data_dir)

+

+ if len(files_to_delete) == 0:

+ console.print(

+ "Nothing to delete. We are safe!", style="green")

+ else:

+ console.print("Here are the files to delete:")

+ for f in files_to_delete:

+ console.print(f, style="magenta")

+ if delete_data_dir:

+ console.print(data_dir + " (data dir)",

+ style="magenta")

+

+ to_delete = Confirm.ask(

+ "[red]Are you sure you want to delete them?[/]", default=False, console=console)

+ if to_delete:

+ for f in files_to_delete:

+ os.remove(f)

+ if delete_data_dir:

+ # recursively delete data dir

+ shutil.rmtree(data_dir)

+ console.print(

+ "Your workspace is clean now!", style="green")

+import os.path

+import re

+

+

+class Template:

+ def __init__(self, template_path: str, var_prefix: str = "CRUPEST"):

+ if len(var_prefix) != 0 and re.fullmatch(r"^[a-zA-Z_][a-zA-Z0-9_]*$", var_prefix) is None:

+ raise ValueError("Invalid var prefix.")

+ self.template_path = template_path

+ self.template_name = os.path.basename(

+ template_path)[:-len(".template")]

+ with open(template_path, "r") as f:

+ self.template = f.read()

+ self.var_prefix = var_prefix

+ self.__var_regex = re.compile(r"\$(" + var_prefix + r"_[a-zA-Z0-9_]+)")

+ self.__var_brace_regex = re.compile(

+ r"\$\{\s*(" + var_prefix + r"_[a-zA-Z0-9_]+)\s*\}")

+ var_set = set()

+ for match in self.__var_regex.finditer(self.template):

+ var_set.add(match.group(1))

+ for match in self.__var_brace_regex.finditer(self.template):

+ var_set.add(match.group(1))

+ self.var_set = var_set

+

+ def generate(self, config: dict) -> str:

+ result = self.template

+ for var in self.var_set:

+ if var not in config:

+ raise ValueError(f"Missing config var {var}.")

+ result = result.replace("$" + var, config[var])

+ result = re.sub(r"\$\{\s*" + var + r"\s*\}", config[var], result)

+ return result

+import os.path

+import re

+

+_template_filename_suffix = ".template"

+_template_var_regex = r"\$([-_a-zA-Z0-9]+)"

+_template_var_brace_regex = r"\$\{\s*([-_a-zA-Z0-9]+?)\s*\}"

+

+

+class Template2:

+

+ @staticmethod

+ def from_file(template_path: str) -> "Template2":

+ if not template_path.endswith(_template_filename_suffix):

+ raise Exception(

+ "Template file must have a name ending with .template.")

+ template_name = os.path.basename(

+ template_path)[:-len(_template_filename_suffix)]

+ with open(template_path, "r") as f:

+ template = f.read()

+ return Template2(template_name, template, template_path=template_path)

+

+ def __init__(self, template_name: str, template: str, *, template_path: str | None = None) -> None:

+ self.template_name = template_name

+ self.template = template

+ self.template_path = template_path

+ self.var_set = set()

+ for match in re.finditer(_template_var_regex, self.template):

+ self.var_set.add(match.group(1))

+ for match in re.finditer(_template_var_brace_regex, self.template):

+ self.var_set.add(match.group(1))

+

+ def partial_render(self, vars: dict[str, str]) -> "Template2":

+ t = self.render(vars)

+ return Template2(self.template_name, t, template_path=self.template_path)

+

+ def render(self, vars: dict[str, str]) -> str:

+ for name in vars.keys():

+ if name not in self.var_set:

+ raise ValueError(f"Invalid var name {name}.")

+

+ text = self.template

+ for name, value in vars.items():

+ text = text.replace("$" + name, value)

+ text = re.sub(r"\$\{\s*" + name + r"\s*\}", value, text)

+ return text

+import json

+from http.client import *

+from urllib.request import urlopen

+

+

+def test_crupest_api(console):

+ def do_the_test():

+ res: HTTPResponse = urlopen("http://localhost:5188/api/todos")

+ body = res.read()

+

+ if res.status != 200:

+ raise Exception("Status code is not 200.")

+ result = json.loads(body)

+ if not isinstance(result, list):

+ raise Exception("Result is not an array.")

+ if len(result) == 0:

+ raise Exception("Result is an empty array.")

+ if not isinstance(result[0], dict):

+ raise Exception("Result[0] is not an object.")

+ if not isinstance(result[0].get("title"), str):

+ raise Exception("Result[0].title is not a string.")

+ if not isinstance(result[0].get("status"), str):

+ raise Exception("Result[0].status is not a string.")

+

+ try:

+ do_the_test()

+ console.print("Test passed!", style="green")

+ exit(0)

+ except Exception as e:

+ console.print(e)

+ console.print("Test failed!", style="red")

+from rich.console import Console

+from rich.prompt import Prompt, Confirm

+

+Prompt = Prompt

+Confirm = Confirm

+

+console = Console()

+from .tui import console

+

+good_style = "green"

+warning_style = "yellow"

+error_style = "red bold"

+file_name_style = "cyan bold"

+var_style = "magenta bold"

+value_style = "cyan bold"

+bye_style = "cyan"

+

+

+def print_with_indent(value: str, style: str, /, indent: int = 0, *, indent_width: int = 2, end='\n'):

+ console.print(

+ f'{" " * indent * indent_width}[{style}]{value}[/]', end=end)

+

+

+def print_var_value(name: str, value: str, /, indent: int = 0, *, indent_width: int = 2, end='\n'):

+ console.print(

+ f'{" " * indent * indent_width}[{var_style}]{name}[/] = [{value_style}]{value}[/]', end=end)

+rich

+jsonschema

+cryptography

+#! /usr/bin/env sh

+exec python3 "$(dirname "$0")/aio.py" update-blog "$@"

+#! /usr/bin/env sh

+

+set -e

+

+cd "$(dirname "$0")/../.."

+

+exec tmux new-session 'cd docker/crupest-nginx/sites/www && pnpm start' \; \

+ split-window -h 'cd docker/crupest-api/CrupestApi/CrupestApi && dotnet run --launch-profile dev'