# 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 immutables import Map from ..versions import VerTuple from ..url_patterns import ParsedPattern class EnabledStatus(Enum): """ ENABLED - User wished to always apply given mapping when it matched. DISABLED - User wished to never apply given mapping. AUTO_ENABLED - User has not configured given mapping but it will still be used when automatic application of mappings is turned on. NO_MARK - User has not configured given mapping and it won't be used. """ ENABLED = 'E' DISABLED = 'D' AUTO_ENABLED = 'A' NO_MARK = 'N' @dc.dataclass(frozen=True, unsafe_hash=True) class Ref: """....""" id: str # 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, state: 'HaketiloState') -> None: """....""" ... @abstractmethod def update( self, state: 'HaketiloState', *, name: t.Optional[str] = None, url: t.Optional[str] = None ) -> 'RepoRef': """....""" ... @abstractmethod def refresh(self, state: 'HaketiloState') -> 'RepoIterationRef': """....""" ... @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class RepoIterationRef(Ref): """....""" pass @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class MappingRef(Ref): """....""" @abstractmethod def disable(self, state: 'HaketiloState') -> None: """....""" ... @abstractmethod def forget_enabled(self, state: 'HaketiloState') -> None: """....""" ... @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class MappingVersionRef(Ref): """....""" @abstractmethod def enable(self, state: 'HaketiloState') -> None: """....""" ... @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class ResourceRef(Ref): """....""" pass @dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc] class ResourceVersionRef(Ref): """....""" pass @dc.dataclass(frozen=True) class PayloadKey: """....""" payload_ref: 'PayloadRef' mapping_identifier: str # mapping_version: VerTuple # mapping_repo: str # mapping_repo_iteration: int pattern: ParsedPattern def __lt__(self, other: 'PayloadKey') -> bool: """....""" return ( self.mapping_identifier, # other.mapping_version, # self.mapping_repo, # other.mapping_repo_iteration, self.pattern ) < ( other.mapping_identifier, # self.mapping_version, # other.mapping_repo, # self.mapping_repo_iteration, other.pattern ) @dc.dataclass(frozen=True) class PayloadData: """....""" payload_ref: 'PayloadRef' mapping_installed: bool explicitly_enabled: bool unique_token: str pattern: ParsedPattern 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, state: 'HaketiloState') -> PayloadData: """....""" ... @abstractmethod def get_mapping(self, state: 'HaketiloState') -> MappingVersionRef: """....""" ... @abstractmethod def get_script_paths(self, state: 'HaketiloState') \ -> t.Iterable[t.Sequence[str]]: """....""" ... @abstractmethod def get_file_data(self, state: 'HaketiloState', 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 get_repo(self, repo_id: str) -> RepoRef: """....""" ... @abstractmethod def get_repo_iteration(self, repo_iteration_id: str) -> RepoIterationRef: """....""" ... @abstractmethod def get_mapping(self, mapping_id: str) -> MappingRef: """....""" ... @abstractmethod def get_mapping_version(self, mapping_version_id: str) \ -> MappingVersionRef: """....""" ... @abstractmethod def get_resource(self, resource_id: str) -> ResourceRef: """....""" ... @abstractmethod def get_resource_version(self, resource_version_id: str) \ -> ResourceVersionRef: """....""" ... @abstractmethod def get_payload(self, payload_id: str) -> PayloadRef: """....""" ... @abstractmethod def add_repo(self, name: t.Optional[str], url: t.Optional[str]) \ -> RepoRef: """....""" ... @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: """....""" ...