# 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 <https://www.gnu.org/licenses/>.
#
#
# I, Wojtek Kosior, thereby promise not to sue for violation of this
# file's license. Although I request that you do not make use of 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.
"""
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 matches site's
URL.
DISABLED - User wished to never apply given mapping.
NO_MARK - User has not configured given mapping.
"""
ENABLED = 'E'
DISABLED = 'D'
NO_MARK = 'N'
class FrozenStatus(Enum):
"""
EXACT_VERSION - User wished to always use the same version of a mapping.
REPOSITORY - User wished to always use a version of the mapping from the
same repository.
NOT_FROZEN - User did not restrict updates of the mapping.
"""
EXACT_VERSION = 'E'
REPOSITORY = 'R'
NOT_FROZEN = 'N'
@staticmethod
def make(letter: t.Optional[str]) -> t.Optional['FrozenStatus']:
if letter is None:
return None
return FrozenStatus(letter)
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 RulePatternInvalid(HaketiloException):
pass
@dc.dataclass(frozen=True)
class RuleDisplayInfo:
ref: 'RuleRef'
pattern: str
allow_scripts: bool
# 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 RuleRef(Ref):
@abstractmethod
def remove(self) -> None:
...
@abstractmethod
def update(
self,
*,
pattern: t.Optional[str] = None,
allow: t.Optional[bool] = None
) -> None:
...
@abstractmethod
def get_display_info(self) -> RuleDisplayInfo:
...
class RuleStore(Store[RuleRef]):
@abstractmethod
def get_display_infos(self, allow: t.Optional[bool] = None) \
-> t.Sequence[RuleDisplayInfo]:
...
@abstractmethod
def add(self, pattern: str, allow: bool) -> RuleRef:
...
@abstractmethod
def get_by_pattern(self, pattern: str) -> RuleRef:
...
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
@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 FileData:
mime_type: str
name: str
contents: bytes
@dc.dataclass(frozen=True)
class MappingDisplayInfo(item_infos.CorrespondsToMappingDCMixin):
ref: 'MappingRef'
identifier: str
enabled: EnabledStatus
frozen: t.Optional[FrozenStatus]
active_version: t.Optional['MappingVersionDisplayInfo']
@dc.dataclass(frozen=True)
class RichMappingDisplayInfo(MappingDisplayInfo):
all_versions: t.Sequence['MappingVersionDisplayInfo']
@dc.dataclass(frozen=True)
class MappingVersionDisplayInfo(item_infos.CorrespondsToMappingDCMixin):
ref: 'MappingVersionRef'
info: item_infos.MappingInfo
installed: InstalledStatus
active: ActiveStatus
is_orphan: bool
is_local: bool
@dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc]
class MappingRef(Ref, item_infos.CorrespondsToMappingDCMixin):
"""...."""
@abstractmethod
def update_status(
self,
enabled: EnabledStatus,
frozen: t.Optional[FrozenStatus] = None
) -> None:
...
@abstractmethod
def get_display_info(self) -> RichMappingDisplayInfo:
...
class MappingStore(Store[MappingRef]):
@abstractmethod
def get_display_infos(self) -> t.Sequence[MappingDisplayInfo]:
...
@abstractmethod
def get_by_identifier(self, identifier: str) -> MappingRef:
...
@dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc]
class MappingVersionRef(Ref, item_infos.CorrespondsToMappingDCMixin):
@abstractmethod
def install(self) -> None:
...
@abstractmethod
def uninstall(self) -> t.Optional['MappingVersionRef']:
...
@abstractmethod
def ensure_depended_items_installed(self) -> None:
...
@abstractmethod
def update_mapping_status(
self,
enabled: EnabledStatus,
frozen: t.Optional[FrozenStatus] = None
) -> None:
...
@abstractmethod
def get_license_file(self, name: str) -> FileData:
...
@abstractmethod
def get_upstream_license_file_url(self, name: str) -> str:
...
@abstractmethod
def get_required_mapping(self, identifier: str) -> 'MappingVersionRef':
...
@abstractmethod
def get_payload_resource(self, pattern: str, identifier: str) \
-> 'ResourceVersionRef':
...
@abstractmethod
def get_item_display_info(self) -> RichMappingDisplayInfo:
...
class MappingVersionStore(Store[MappingVersionRef]):
pass
@dc.dataclass(frozen=True)
class ResourceDisplayInfo(item_infos.CorrespondsToResourceDCMixin):
ref: 'ResourceRef'
identifier: str
@dc.dataclass(frozen=True)
class RichResourceDisplayInfo(ResourceDisplayInfo):
all_versions: t.Sequence['ResourceVersionDisplayInfo']
@dc.dataclass(frozen=True)
class ResourceVersionDisplayInfo(item_infos.CorrespondsToResourceDCMixin):
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, item_infos.CorrespondsToResourceDCMixin):
@abstractmethod
def get_display_info(self) -> RichResourceDisplayInfo:
...
class ResourceStore(Store[ResourceRef]):
@abstractmethod
def get_display_infos(self) -> t.Sequence[ResourceDisplayInfo]:
...
@abstractmethod
def get_by_identifier(self, identifier: str) -> ResourceRef:
...
@dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc]
class ResourceVersionRef(Ref, item_infos.CorrespondsToResourceDCMixin):
@abstractmethod
def install(self) -> None:
...
@abstractmethod
def uninstall(self) -> t.Optional['ResourceVersionRef']:
...
@abstractmethod
def get_license_file(self, name: str) -> FileData:
...
@abstractmethod
def get_resource_file(self, name: str) -> FileData:
...
@abstractmethod
def get_upstream_license_file_url(self, name: str) -> str:
...
@abstractmethod
def get_upstream_resource_file_url(self, name: str) -> str:
...
@abstractmethod
def get_dependency(self, identifier: str) -> 'ResourceVersionRef':
...
@abstractmethod
def get_item_display_info(self) -> RichResourceDisplayInfo:
...
class ResourceVersionStore(Store[ResourceVersionRef]):
pass
@dc.dataclass(frozen=True)
class PayloadKey:
"""...."""
ref: 'PayloadRef'
mapping_identifier: str
def __lt__(self, other: 'PayloadKey') -> bool:
"""...."""
return self.mapping_identifier < other.mapping_identifier
@dc.dataclass(frozen=True)
class PayloadData:
"""...."""
ref: 'PayloadRef'
explicitly_enabled: bool
unique_token: str
mapping_identifier: str
pattern: str
pattern_path_segments: tuple[str, ...]
eval_allowed: bool
cors_bypass_allowed: bool
global_secret: bytes
@dc.dataclass(frozen=True)
class PayloadDisplayInfo:
ref: 'PayloadRef'
mapping_info: MappingVersionDisplayInfo
pattern: str
has_problems: bool
@dc.dataclass(frozen=True, unsafe_hash=True) # type: ignore[misc]
class PayloadRef(Ref):
"""...."""
@abstractmethod
def get_data(self) -> PayloadData:
"""...."""
...
@abstractmethod
def has_problems(self) -> bool:
...
@abstractmethod
def get_display_info(self) -> PayloadDisplayInfo:
...
@abstractmethod
def ensure_items_installed(self) -> None:
"""...."""
...
@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 PayloadStore(Store[PayloadRef]):
pass
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'
class PopupStyle(Enum):
"""
DIALOG - Make popup open inside an iframe on the current page.
TAB - Make popup open in a new tab.
"""
DIALOG = 'D'
TAB = 'T'
@dc.dataclass(frozen=True)
class PopupSettings:
# We'll implement button later.
#button_trigger: bool
keyboard_trigger: bool
style: PopupStyle
@property
def popup_enabled(self) -> bool:
return self.keyboard_trigger #or self.button_trigger
@dc.dataclass(frozen=True)
class HaketiloGlobalSettings:
"""...."""
mapping_use_mode: MappingUseMode
default_allow_scripts: bool
advanced_user: bool
repo_refresh_seconds: int
locale: t.Optional[str]
update_waiting: bool
default_popup_jsallowed: PopupSettings
default_popup_jsblocked: PopupSettings
default_popup_payloadon: PopupSettings
class Logger(ABC):
@abstractmethod
def warn(self, msg: str) -> None:
...
class MissingItemError(ValueError):
"""...."""
pass
@dc.dataclass(frozen=True)
class OrphanItemsStats:
mappings: int
resources: int
class HaketiloState(ABC):
"""...."""
@abstractmethod
def import_items(self, malcontent_path: Path) -> None:
...
@abstractmethod
def count_orphan_items(self) -> OrphanItemsStats:
...
@abstractmethod
def prune_orphan_items(self) -> None:
...
@abstractmethod
def rule_store(self) -> RuleStore:
...
@abstractmethod
def repo_store(self) -> RepoStore:
"""...."""
...
@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 payload_store(self) -> PayloadStore:
...
@abstractmethod
def get_secret(self) -> bytes:
...
@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,
advanced_user: t.Optional[bool] = None,
repo_refresh_seconds: t.Optional[int] = None,
locale: t.Optional[str] = None,
default_popup_settings: t.Mapping[str, PopupSettings] = {}
) -> None:
...
@abstractmethod
def upate_all_items(self) -> None:
...
@property
@abstractmethod
def listen_host(self) -> str:
...
@property
@abstractmethod
def listen_port(self) -> int:
...
@abstractmethod
def launch_browser(self) -> bool:
...
@property
@abstractmethod
def logger(self) -> Logger:
...
