diff options
Diffstat (limited to 'src/hydrilla/proxy/policies/base.py')
| -rw-r--r-- | src/hydrilla/proxy/policies/base.py | 177 |
1 files changed, 177 insertions, 0 deletions
diff --git a/src/hydrilla/proxy/policies/base.py b/src/hydrilla/proxy/policies/base.py new file mode 100644 index 0000000..3bde6f2 --- /dev/null +++ b/src/hydrilla/proxy/policies/base.py @@ -0,0 +1,177 @@ +# 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 <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 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)) |