# SPDX-License-Identifier: GPL-3.0-or-later # Base defintions for policies for altering HTTP requests. # # 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. """ ..... """ # Enable using with Python 3.7. from __future__ import annotations import sys if sys.version_info >= (3, 8): from typing import Protocol else: from typing_extensions import Protocol import dataclasses as dc import typing as t import enum from abc import ABC, abstractmethod from immutables import Map from ...url_patterns import ParsedUrl from .. import state class PolicyPriority(int, enum.Enum): """....""" _ONE = 1 _TWO = 2 _THREE = 3 DefaultGetValue = t.TypeVar('DefaultGetValue', object, None) class IHeaders(Protocol): """....""" def __getitem__(self, key: str) -> str: ... def get_all(self, key: str) -> t.Iterable[str]: ... def get(self, key: str, default: DefaultGetValue = None) \ -> t.Union[str, DefaultGetValue]: ... def items(self) -> t.Iterable[tuple[str, str]]: ... def encode_headers_items(headers: t.Iterable[tuple[str, str]]) \ -> t.Iterable[tuple[bytes, bytes]]: """....""" for name, value in headers: yield name.encode(), value.encode() @dc.dataclass(frozen=True) class ProducedRequest: """....""" url: str method: str headers: t.Iterable[tuple[bytes, bytes]] body: bytes @dc.dataclass(frozen=True) class RequestInfo: """....""" url: ParsedUrl method: str headers: IHeaders body: bytes def make_produced_request(self) -> ProducedRequest: """....""" return ProducedRequest( url = self.url.orig_url, method = self.method, headers = encode_headers_items(self.headers.items()), body = self.body ) @dc.dataclass(frozen=True) class ProducedResponse: """....""" status_code: int headers: t.Iterable[tuple[bytes, bytes]] body: bytes @dc.dataclass(frozen=True) class ResponseInfo: """....""" url: ParsedUrl status_code: int headers: IHeaders body: bytes def make_produced_response(self) -> ProducedResponse: """....""" return ProducedResponse( status_code = self.status_code, headers = encode_headers_items(self.headers.items()), body = self.body ) class Policy(ABC): """....""" process_request: t.ClassVar[bool] = False process_response: t.ClassVar[bool] = False priority: t.ClassVar[PolicyPriority] @property def anticache(self) -> bool: return self.process_request or self.process_response def consume_request(self, request_info: RequestInfo) \ -> t.Optional[t.Union[ProducedRequest, ProducedResponse]]: """....""" return None def consume_response(self, response_info: ResponseInfo) \ -> t.Optional[ProducedResponse]: """....""" return None # 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 PolicyFactory(ABC): """....""" builtin: bool @abstractmethod def make_policy(self, haketilo_state: state.HaketiloState) \ -> t.Optional[Policy]: """....""" ... def __lt__(self, other: 'PolicyFactory'): """....""" return sorting_keys.get(self.__class__.__name__, 999) < \ sorting_keys.get(other.__class__.__name__, 999) sorting_order = ( 'PayloadResourcePolicyFactory', 'PayloadPolicyFactory', 'RuleBlockPolicyFactory', 'RuleAllowPolicyFactory', 'FallbackPolicyFactory' ) sorting_keys = Map((cls, name) for name, cls in enumerate(sorting_order))