Module ogr.services.github.auth_providers

class GithubApp(GithubAuthentication):
    def __init__(self, id: str, private_key: str, private_key_path: str) -> None:
        self.id = id
        self._private_key = private_key
        self._private_key_path = private_key_path

        self._github: github.Github = None
        self._integration: github.GithubIntegration = None

    def __eq__(self, o: object) -> bool:
        if not issubclass(o.__class__, GithubApp):
            return False

        return (
            self.id == o.id  # type: ignore
            and self._private_key == o._private_key  # type: ignore
            and self._private_key_path == o._private_key_path  # type: ignore
        )

    def __str__(self) -> str:
        censored_id = f"id='{self.id[:1]}***{self.id[-1:]}'"
        censored_private_key = (
            f", private_key" f"='{self._private_key[:1]}***{self._private_key[-1:]}'"
            if self._private_key
            else ""
        )
        private_key_path = (
            f", private_key_path='{self._private_key_path}'"
            if self._private_key_path
            else ""
        )

        return f"GithubApp({censored_id}{censored_private_key}{private_key_path})"

    @property
    def private_key(self) -> str:
        if self._private_key:
            return self._private_key

        if self._private_key_path:
            if not Path(self._private_key_path).is_file():
                raise OgrException(
                    f"File with the github-app private key "
                    f"({self._private_key_path}) "
                    f"does not exist.",
                )
            return Path(self._private_key_path).read_text()

        return None

    @property
    def pygithub_instance(self) -> Optional[github.Github]:
        # used for backward compatibility with GitUser
        return None

    @property
    def integration(self) -> github.GithubIntegration:
        if not self._integration:
            self._integration = github.GithubIntegration(self.id, self.private_key)
        return self._integration

    def get_token(self, namespace: str, repo: str) -> str:
        if not self.private_key:
            return None

        # PyGithub 1.58 deprecated get_installation() in favor of get_repo_installation()
        # that raises an exception on error rather than returning None
        if hasattr(self.integration, "get_repo_installation"):
            try:
                inst_id = self.integration.get_repo_installation(namespace, repo).id
            except github.GithubException:
                inst_id = None
        else:
            inst_id = self.integration.get_installation(namespace, repo).id
        # PyGithub<1.52 returned an object for id, with a value attribute,
        # which was None or an ID.
        # This was changed in:
        # https://github.com/PyGithub/PyGithub/commit/61808da15e8e3bcb660acd0e7947326a4a6c0c7a#diff-b8f1ee87df332916352809a397ea259aL54
        # 'id' is now None or an ID.
        inst_id = (
            inst_id
            if isinstance(inst_id, int) or inst_id is None
            else inst_id.value  # type: ignore
        )
        if not inst_id:
            raise OgrException(
                f"No installation ID provided for {namespace}/{repo}: "
                "please make sure that you provided correct credentials of your GitHub app.",
            )
        inst_auth = self.integration.get_access_token(inst_id)  # type: ignore
        return inst_auth.token

    @staticmethod
    def try_create(
        github_app_id: Optional[str] = None,
        github_app_private_key: Optional[str] = None,
        github_app_private_key_path: Optional[str] = None,
        **_,
    ) -> Optional["GithubApp"]:
        return (
            GithubApp(
                github_app_id,
                github_app_private_key,
                github_app_private_key_path,
            )
            if github_app_id
            else None
        )

class GithubAuthentication:
    """
    Represents a token manager for authentication via GitHub App.
    """

    def get_token(self, namespace: str, repo: str) -> str:
        """
        Get a GitHub token for requested repository.

        Args:
            namespace: Namespace of the repository.
            repo: Name of the repository.

        Returns:
            A token that can be used in PyGithub instance for authentication.
        """
        raise NotImplementedError()

    @property
    def pygithub_instance(self) -> "github.Github":
        """
        Returns:
            Generic PyGithub instance. Used for `GitUser` for example.
        """
        raise NotImplementedError()

    @staticmethod
    def try_create(**kwargs) -> Optional["GithubAuthentication"]:
        """
        Tries to construct authentication object from provided keyword arguments.

        Returns:
            `GithubAuthentication` object or `None` if the creation was not
            successful.
        """
        raise NotImplementedError()

class TokenAuthentication(GithubAuthentication):
    def __init__(self, token: str, max_retries: Union[int, Retry] = 0, **_) -> None:
        self._token = token
        self._pygithub_instance = github.Github(login_or_token=token, retry=max_retries)

    def __eq__(self, o: object) -> bool:
        return issubclass(o.__class__, TokenAuthentication) and (
            self._token == o._token  # type: ignore
        )

    def __str__(self) -> str:
        censored_token = (
            f"token='{self._token[:1]}***{self._token[-1:]}'" if self._token else ""
        )
        return f"Token({censored_token})"

    @property
    def pygithub_instance(self) -> github.Github:
        return self._pygithub_instance

    def get_token(self, namespace: str, repo: str) -> str:
        return self._token

    @staticmethod
    def try_create(
        token: Optional[str] = None,
        max_retries: Union[int, Retry] = 0,
        **_,
    ) -> Optional["TokenAuthentication"]:
        return TokenAuthentication(token, max_retries=max_retries)

class Tokman(GithubAuthentication):
    def __init__(self, instance_url: str):
        self._instance_url = instance_url

    def __eq__(self, o: object) -> bool:
        if not issubclass(o.__class__, Tokman):
            return False

        return self._instance_url == o._instance_url  # type: ignore

    def __str__(self) -> str:
        return f"Tokman(instance_url='{self._instance_url}')"

    @property
    def pygithub_instance(self) -> Optional[github.Github]:
        # used for backward compatibility with GitUser
        return None

    def get_token(self, namespace: str, repo: str) -> str:
        response = requests.get(f"{self._instance_url}/api/{namespace}/{repo}")

        if not response.ok:
            if response.status_code == 400:
                raise GithubAppNotInstalledError(response.text)

            cls = OgrNetworkError if response.status_code >= 500 else OgrException
            raise cls(
                f"Couldn't retrieve token from Tokman: ({response.status_code}) {response.text}",
            )

        return response.json().get("access_token", None)

    @staticmethod
    def try_create(
        tokman_instance_url: Optional[str] = None,
        **_,
    ) -> Optional["Tokman"]:
        return Tokman(tokman_instance_url) if tokman_instance_url else None