# SPDX-License-Identifier: GPL-3.0-or-later # Haketilo proxy data and configuration (interface definition through abstract # class). # # This file is part of Hydrilla&Haketilo. # # Copyright (C) 2022 Wojtek Kosior # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . # # # I, Wojtek Kosior, thereby promise not to sue for violation of this # file's license. Although I request that you do not make use this code # in a proprietary program, I am not going to enforce this in court. """ This module defines API for keeping track of all settings, rules, mappings and resources. """ # Enable using with Python 3.7. from __future__ import annotations import dataclasses as dc import typing as t from pathlib import Path from abc import ABC, abstractmethod from enum import Enum from datetime import datetime from immutables import Map from ..exceptions import HaketiloException from ..versions import VerTuple from ..url_patterns import ParsedPattern from .. import item_infos from .simple_dependency_satisfying import ImpossibleSituation class EnabledStatus(Enum): """ ENABLED - User wished to always apply given mapping when it matched. DISABLED - User wished to never apply given mapping. NO_MARK - User has not configured given mapping and it won't be used. """ ENABLED = 'E' DISABLED = 'D' NO_MARK = 'N' class InstalledStatus(Enum): """ INSTALLED - Mapping's all files are present and mapping data is not going to be automatically removed. NOT_INSTALLED - Some of the mapping's files might be absent. Mapping can be automatically removed if it is orphaned. FAILED_TO_INSTALL - Same as "NOT_INSTALLED" but we additionally know that the last automatic attempt to install mapping's files from repository was unsuccessful. """ INSTALLED = 'I' NOT_INSTALLED = 'N' FAILED_TO_INSTALL = 'F' class ActiveStatus(Enum): """ REQUIRED - Mapping version got active to fulfill a requirement of some (this or another) explicitly enabled mapping. AUTO - Mapping version was activated automatically. NOT_ACTIVE - Mapping version is not currently being used. """ REQUIRED = 'R' AUTO = 'A' NOT_ACTIVE = 'N' @dc.dataclass(frozen=True, unsafe_hash=True) class Ref: """....""" id: str def __post_init__(self): assert isinstance(self.id, str) RefType = t.TypeVar('RefType', bound=Ref) class Store(ABC, t.Generic[RefType]): @abstractmethod def get(self, id) -> RefType: ... class RepoNameInvalid(HaketiloException): pass class RepoNameTaken(HaketiloException): pass class RepoUrlInvalid(HaketiloException): pass class RepoCommunicationError(HaketiloException): pass @dc.dataclass(frozen=True) class FileInstallationError(HaketiloException): repo_id: str sha256: str @dc.dataclass(frozen=True) class FileIntegrityError(FileInstallationError): invalid_sha256: str @dc.dataclass(frozen=True) class FileMissingError(FileInstallationError): pass class RepoApiVersionUnsupported(HaketiloException): pass # mypy needs to be corrected: # https://stackoverflow.com/questions/70999513/conflict-between-mix-ins-for-abstract-dataclasses/70999704#70999704 @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class RepoRef(Ref): """....""" @abstractmethod def remove(self) -> None: """....""" ... @abstractmethod def update( self, *, name: t.Optional[str] = None, url: t.Optional[str] = None ) -> None: """....""" ... @abstractmethod def refresh(self) -> None: """....""" ... @abstractmethod def get_display_info(self) -> 'RepoDisplayInfo': ... @dc.dataclass(frozen=True) class RepoDisplayInfo: ref: RepoRef is_local_semirepo: bool name: str url: str deleted: bool last_refreshed: t.Optional[datetime] resource_count: int mapping_count: int class RepoStore(Store[RepoRef]): @abstractmethod def get_display_infos(self, include_deleted: bool = False) -> \ t.Sequence[RepoDisplayInfo]: ... @abstractmethod def add(self, name: str, url: str) -> RepoRef: ... @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class RepoIterationRef(Ref): """....""" pass @dc.dataclass(frozen=True) class MappingDisplayInfo: ref: 'MappingRef' identifier: str enabled: EnabledStatus active_version: t.Optional['MappingVersionDisplayInfo'] @dc.dataclass(frozen=True) class MappingVersionDisplayInfo: ref: 'MappingVersionRef' info: item_infos.MappingInfo installed: InstalledStatus active: ActiveStatus is_orphan: bool is_local: bool mapping_enabled: EnabledStatus @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class MappingRef(Ref): """....""" @abstractmethod def get_version_display_infos(self) \ -> t.Sequence[MappingVersionDisplayInfo]: ... class MappingStore(Store[MappingRef]): @abstractmethod def get_display_infos(self) -> t.Sequence[MappingDisplayInfo]: ... @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class MappingVersionRef(Ref): @abstractmethod def install(self) -> None: ... @abstractmethod def uninstall(self) -> t.Optional['MappingVersionRef']: ... @abstractmethod def get_all_version_display_infos(self) \ -> t.Sequence[MappingVersionDisplayInfo]: ... class MappingVersionStore(Store[MappingVersionRef]): pass @dc.dataclass(frozen=True) class ResourceDisplayInfo: ref: 'ResourceRef' identifier: str @dc.dataclass(frozen=True) class ResourceVersionDisplayInfo: ref: 'ResourceVersionRef' info: item_infos.ResourceInfo installed: InstalledStatus active: ActiveStatus is_orphan: bool is_local: bool @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class ResourceRef(Ref): @abstractmethod def get_version_display_infos(self) \ -> t.Sequence[ResourceVersionDisplayInfo]: ... class ResourceStore(Store[ResourceRef]): @abstractmethod def get_display_infos(self) -> t.Sequence[ResourceDisplayInfo]: ... @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class ResourceVersionRef(Ref): @abstractmethod def install(self) -> None: ... @abstractmethod def uninstall(self) -> t.Optional['ResourceVersionRef']: ... @abstractmethod def get_all_version_display_infos(self) \ -> t.Sequence[ResourceVersionDisplayInfo]: ... class ResourceVersionStore(Store[ResourceVersionRef]): pass @dc.dataclass(frozen=True) class PayloadKey: """....""" payload_ref: 'PayloadRef' mapping_identifier: str def __lt__(self, other: 'PayloadKey') -> bool: """....""" return self.mapping_identifier < other.mapping_identifier @dc.dataclass(frozen=True) class PayloadData: """....""" payload_ref: 'PayloadRef' explicitly_enabled: bool unique_token: str pattern_path_segments: tuple[str, ...] eval_allowed: bool cors_bypass_allowed: bool @dc.dataclass(frozen=True) class FileData: type: str name: str contents: bytes @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class PayloadRef(Ref): """....""" @abstractmethod def get_data(self) -> PayloadData: """....""" ... @abstractmethod def get_mapping(self) -> MappingVersionRef: """....""" ... @abstractmethod def get_script_paths(self) \ -> t.Iterable[t.Sequence[str]]: """....""" ... @abstractmethod def get_file_data(self, path: t.Sequence[str]) \ -> t.Optional[FileData]: """....""" ... class MappingUseMode(Enum): """ AUTO - Apply mappings except for those explicitly disabled. WHEN_ENABLED - Only apply mappings explicitly marked as enabled. Don't apply unmarked nor explicitly disabled mappings. QUESTION - Automatically apply mappings that are explicitly enabled. Ask whether to enable unmarked mappings. Don't apply explicitly disabled ones. """ AUTO = 'A' WHEN_ENABLED = 'W' QUESTION = 'Q' @dc.dataclass(frozen=True) class HaketiloGlobalSettings: """....""" mapping_use_mode: MappingUseMode default_allow_scripts: bool repo_refresh_seconds: int class MissingItemError(ValueError): """....""" pass class HaketiloState(ABC): """....""" @abstractmethod def import_items(self, malcontent_path: Path) -> None: ... @abstractmethod def repo_store(self) -> RepoStore: """....""" ... @abstractmethod def get_repo_iteration(self, repo_iteration_id: str) -> RepoIterationRef: """....""" ... @abstractmethod def mapping_store(self) -> MappingStore: ... @abstractmethod def mapping_version_store(self) -> MappingVersionStore: ... @abstractmethod def resource_store(self) -> ResourceStore: ... @abstractmethod def resource_version_store(self) -> ResourceVersionStore: ... @abstractmethod def get_payload(self, payload_id: str) -> PayloadRef: """....""" ... @abstractmethod def get_settings(self) -> HaketiloGlobalSettings: """....""" ... @abstractmethod def update_settings( self, *, mapping_use_mode: t.Optional[MappingUseMode] = None, default_allow_scripts: t.Optional[bool] = None, repo_refresh_seconds: t.Optional[int] = None ) -> None: """....""" ...