diff options
Diffstat (limited to 'tools/cru-py/cru/service/nginx.py')
| -rw-r--r-- | tools/cru-py/cru/service/nginx.py | 356 |
1 files changed, 0 insertions, 356 deletions
diff --git a/tools/cru-py/cru/service/nginx.py b/tools/cru-py/cru/service/nginx.py index 94c4375..ad32cb9 100644 --- a/tools/cru-py/cru/service/nginx.py +++ b/tools/cru-py/cru/service/nginx.py @@ -4,7 +4,6 @@ import re import subprocess from typing import Literal, Any, cast, ClassVar -import jsonschema def restart_nginx(force=False) -> bool: @@ -16,358 +15,3 @@ def restart_nginx(force=False) -> bool: 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") |