# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""
A fake SFTP server for testing.

Do not use this in production!  It has only the features required for the
tests that currently use it, and no security.
"""

import io
import os
import socket
import time as tm
from threading import Event, Thread
from typing import Self, override

from cryptography.hazmat.primitives import asymmetric, serialization
from paramiko import (
    Ed25519Key,
    PKey,
    SFTPAttributes,
    SFTPHandle,
    SFTPServer,
    SFTPServerInterface,
    ServerInterface,
    Transport,
)
from paramiko.common import (
    AUTH_FAILED,
    AUTH_SUCCESSFUL,
    OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED,
    OPEN_SUCCEEDED,
)
from paramiko.sftp import SFTP_FAILURE, SFTP_OK


class FakeSSHServer(ServerInterface):
    """A fake SSH server for testing."""

    def __init__(self, user_key: Ed25519Key) -> None:
        """Construct the server interface with a single permitted user key."""
        self.user_key = user_key

    @override
    def check_channel_request(self, kind: str, chanid: int) -> int:
        """Check if a channel request is granted."""
        chanid  # fake usage for vulture
        return (
            OPEN_SUCCEEDED
            if kind in {"session", "subsystem"}
            else OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
        )

    @override
    def get_allowed_auths(
        self,
        username: str,  # noqa: ARG002, U100
    ) -> str:  # pragma: no cover
        """Return supported authentication methods."""
        return "publickey"

    @override
    def check_auth_publickey(
        self,
        username: str,  # noqa: ARG002, U100
        key: PKey,
    ) -> int:
        """Determine if a client key is acceptable for authentication."""
        return AUTH_SUCCESSFUL if key == self.user_key else AUTH_FAILED


class FakeSFTPServer(SFTPServerInterface):
    """A fake SFTP subsystem for testing."""

    @override
    def open(
        self, path: str, flags: int, attr: SFTPAttributes | None
    ) -> SFTPHandle | int:
        """Open a file on the server and create a handle."""
        assert flags & os.O_WRONLY
        assert flags & os.O_CREAT
        assert attr is not None
        SFTPServer.set_file_attr(path, attr)
        # This will be closed by SFTPHandle.close.
        f = open(path, "wb")  # noqa: SIM115
        handle = SFTPHandle(flags)
        handle.writefile = f  # type: ignore[attr-defined]
        return handle

    @override
    def stat(self, path: str) -> SFTPAttributes | int:
        """Return attributes of a path on the server."""
        try:
            return SFTPAttributes.from_stat(os.stat(path))
        except OSError as e:
            return (
                SFTPServer.convert_errno(e.errno)
                if e.errno is not None
                else SFTP_FAILURE
            )

    @override
    def chattr(self, path: str, attr: SFTPAttributes) -> int:
        """Change the attributes of a file."""
        SFTPServer.set_file_attr(path, attr)
        return SFTP_OK


class WrappedEd25519Key:
    """A wrapper for the various formats of Ed25519 key that we need."""

    def __init__(
        self, private_key: asymmetric.ed25519.Ed25519PrivateKey
    ) -> None:
        """Construct a wrapped key object."""
        self.private_key = private_key

    @classmethod
    def generate(cls) -> Self:
        """Generate an Ed25519 private key."""
        return cls(asymmetric.ed25519.Ed25519PrivateKey.generate())

    @property
    def private_openssh(self) -> str:
        """Return the OpenSSH private key representation of this key."""
        return self.private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.OpenSSH,
            encryption_algorithm=serialization.NoEncryption(),
        ).decode()

    @property
    def public_openssh(self) -> str:  # pragma: no cover
        """Return the OpenSSH public key representation of this key."""
        return (
            self.private_key.public_key()
            .public_bytes(
                encoding=serialization.Encoding.OpenSSH,
                format=serialization.PublicFormat.OpenSSH,
            )
            .decode()
        )

    @property
    def public_paramiko(self) -> Ed25519Key:
        """Return the Paramiko public key representation of this key."""
        return Ed25519Key.from_private_key(io.StringIO(self.private_openssh))


class FakeSFTPServerThread(Thread):
    """Thread that runs a fake SFTP server for testing."""

    def __init__(self) -> None:
        """Initialize the server and start listening."""
        super().__init__()
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.socket.bind(("", 0))
        self.socket.listen()
        self.host_key = WrappedEd25519Key.generate()
        self.user_key = WrappedEd25519Key.generate()
        self.stop_event = Event()

    @property
    def address(self) -> tuple[str, int]:
        """Return the (host, port) to be used by clients."""
        host, port = self.socket.getsockname()
        assert isinstance(host, str)
        assert isinstance(port, int)
        return host, port

    @override
    def run(self) -> None:
        """Run the server."""
        while not self.stop_event.is_set():
            self.socket.settimeout(1)
            try:
                conn, _ = self.socket.accept()
            except TimeoutError:  # pragma: no cover
                continue
            self.socket.settimeout(None)
            transport = Transport(conn)
            transport.add_server_key(self.host_key.public_paramiko)
            transport.set_subsystem_handler("sftp", SFTPServer, FakeSFTPServer)
            server = FakeSSHServer(self.user_key.public_paramiko)
            transport.start_server(server=server)
            channel = transport.accept()
            if channel is not None:  # pragma: no cover
                try:
                    while transport.is_active():
                        tm.sleep(0.1)
                finally:
                    transport.close()
                    channel.close()

    def stop(self) -> None:
        """Stop the server."""
        self.stop_event.set()
        self.join()
        self.socket.close()
