path: root/src/hydrilla/proxy/state_impl/concrete_state.py

# SPDX-License-Identifier: GPL-3.0-or-later

# Haketilo proxy data and configuration (instantiatable HaketiloState subtype).
#
# 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.

"""
This module contains logic for keeping track of all settings, rules, mappings
and resources.
"""

# Enable using with Python 3.7.
from __future__ import annotations

import sqlite3
import secrets
import typing as t
import dataclasses as dc

from pathlib import Path

from ...exceptions import HaketiloException
from ...translations import smart_gettext as _
from ... import url_patterns
from ... import item_infos
from .. import state as st
from .. import policies
from .. import simple_dependency_satisfying as sds
from . import base
from . import items
from . import repos
from . import payloads
from . import _operations


here = Path(__file__).resolve().parent


@dc.dataclass(frozen=True, unsafe_hash=True)
class ConcreteRepoIterationRef(st.RepoIterationRef):
    pass


@dc.dataclass
class ConcreteHaketiloState(base.HaketiloStateWithFields):
    def __post_init__(self) -> None:
        sqlite3.enable_callback_tracebacks(True)

        self._prepare_database()

        self.rebuild_structures()

    def _prepare_database(self) -> None:
        """...."""
        cursor = self.connection.cursor()

        try:
            cursor.execute(
                '''
                SELECT
                        COUNT(name)
                FROM
                        sqlite_master
                WHERE
                        name = 'general' AND type = 'table';
                '''
            )

            (db_initialized,), = cursor.fetchall()

            if not db_initialized:
                cursor.executescript((here.parent / 'tables.sql').read_text())

            else:
                cursor.execute(
                    '''
                    SELECT
                            haketilo_version
                    FROM
                            general;
                    '''
                )

                (db_haketilo_version,) = cursor.fetchone()
                if db_haketilo_version != '3.0b1':
                    raise HaketiloException(_('err.proxy.unknown_db_schema'))

            cursor.execute('PRAGMA FOREIGN_KEYS;')
            if cursor.fetchall() == []:
                raise HaketiloException(_('err.proxy.no_sqlite_foreign_keys'))

            cursor.execute('PRAGMA FOREIGN_KEYS=ON;')
        finally:
            cursor.close()

    def import_items(self, malcontent_path: Path, repo_id: int = 1) -> None:
        with self.cursor(transaction=(repo_id == 1)) as cursor:
            # This method without the repo_id argument exposed is part of the
            # state API. As such, calls with repo_id = 1 (imports of local
            # semirepo packages) create a new transaction. Calls with different
            # values of repo_id are assumed to originate from within the state
            # implementation code and expect an existing transaction. Here, we
            # verify the transaction is indeed present.
            assert self.connection.in_transaction

            _operations._load_packages_no_state_update(
                cursor          = cursor,
                malcontent_path = malcontent_path,
                repo_id         = repo_id
            )

            self.rebuild_structures()

    def prune_orphans(self) -> None:
        with self.cursor() as cursor:
            assert self.connection.in_transaction

            _operations.prune_orphans(cursor)

    def recompute_dependencies(
            self,
            unlocked_required_mappings: base.NoLockArg = []
    ) -> None:
        with self.cursor() as cursor:
            assert self.connection.in_transaction

            _operations._recompute_dependencies_no_state_update(
                cursor                     = cursor,
                unlocked_required_mappings = unlocked_required_mappings
            )

            self.rebuild_structures()

    def pull_missing_files(self) -> None:
        with self.cursor() as cursor:
            assert self.connection.in_transaction

            _operations.pull_missing_files(cursor)

    def rebuild_structures(self) -> None:
        with self.cursor(transaction=True) as cursor:
            cursor.execute(
                '''
                SELECT
                        p.payload_id, p.pattern, p.eval_allowed,
                                p.cors_bypass_allowed,
                        ms.enabled,
                        i.identifier
                FROM
                             payloads         AS p
                        JOIN item_versions    AS iv
                                ON p.mapping_item_id = iv.item_version_id
                        JOIN items            AS i  USING (item_id)
                        JOIN mapping_statuses AS ms USING (item_id);
                '''
            )

        rows = cursor.fetchall()

        new_policy_tree = base.PolicyTree()

        ui_factory = policies.WebUIPolicyFactory(builtin=True)
        web_ui_pattern = 'http*://hkt.mitm.it/***'
        for parsed_pattern in url_patterns.parse_pattern(web_ui_pattern):
            new_policy_tree = new_policy_tree.register(
                parsed_pattern,
                ui_factory
            )

        new_payloads_data: dict[st.PayloadRef, st.PayloadData] = {}

        for row in rows:
            (payload_id_int, pattern, eval_allowed, cors_bypass_allowed,
             enabled_status,
             identifier) = row

            payload_ref = payloads.ConcretePayloadRef(str(payload_id_int), self)

            previous_data = self.payloads_data.get(payload_ref)
            if previous_data is not None:
                token = previous_data.unique_token
            else:
                token = secrets.token_urlsafe(8)

            payload_key = st.PayloadKey(payload_ref, identifier)

            for parsed_pattern in url_patterns.parse_pattern(pattern):
                new_policy_tree = new_policy_tree.register_payload(
                    parsed_pattern,
                    payload_key,
                    token
                )

                pattern_path_segments = parsed_pattern.path_segments

            payload_data = st.PayloadData(
                payload_ref           = payload_ref,
                explicitly_enabled    = enabled_status == 'E',
                unique_token          = token,
                pattern_path_segments = pattern_path_segments,
                eval_allowed          = eval_allowed,
                cors_bypass_allowed   = cors_bypass_allowed
            )

            new_payloads_data[payload_ref] = payload_data

        self.policy_tree   = new_policy_tree
        self.payloads_data = new_payloads_data

    def repo_store(self) -> st.RepoStore:
        return repos.ConcreteRepoStore(self)

    def get_repo_iteration(self, repo_iteration_id: str) -> st.RepoIterationRef:
        return ConcreteRepoIterationRef(repo_iteration_id)

    def mapping_store(self) -> st.MappingStore:
        return items.ConcreteMappingStore(self)

    def mapping_version_store(self) -> st.MappingVersionStore:
        return items.ConcreteMappingVersionStore(self)

    def resource_store(self) -> st.ResourceStore:
        return items.ConcreteResourceStore(self)

    def resource_version_store(self) -> st.ResourceVersionStore:
        return items.ConcreteResourceVersionStore(self)

    def get_payload(self, payload_id: str) -> st.PayloadRef:
        raise NotImplementedError()

    def get_settings(self) -> st.HaketiloGlobalSettings:
        return st.HaketiloGlobalSettings(
            mapping_use_mode       = st.MappingUseMode.AUTO,
            default_allow_scripts  = True,
            repo_refresh_seconds   = 0
        )

    def update_settings(
            self,
            *,
            mapping_use_mode:      t.Optional[st.MappingUseMode] = None,
            default_allow_scripts: t.Optional[bool]              = None,
            repo_refresh_seconds:  t.Optional[int]               = None
    ) -> None:
        raise NotImplementedError()

    @staticmethod
    def make(store_dir: Path) -> 'ConcreteHaketiloState':
        connection = sqlite3.connect(
            str(store_dir / 'sqlite3.db'),
            isolation_level   = None,
            check_same_thread = False
        )
        return ConcreteHaketiloState(
            store_dir      = store_dir,
            connection     = connection
        )