src/fakesshagent/events.py (c0052c7) - fakesshagent.git

events.py

# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info>
#
# SPDX-License-Identifier: Zlib

"""Events from the SSH agent protocol.

As per the Sans-I/O paradigm, the SSH agent protocol stack translates
protocol traffic into a bidirectional stream of events.

"""

from __future__ import annotations

import abc
import dataclasses
import enum
from typing import TYPE_CHECKING, Union

from fakesshagent import _types

if TYPE_CHECKING:
    from collections.abc import Set as AbstractSet
    from typing import ClassVar

from typing_extensions import Literal

EVENT = dataclasses.dataclass(frozen=True)

@EVENT
class AbstractServerEvent(abc.ABC):
    """Base class for events the protocol stack handles internally.

Includes parsed client requests, server responses and automatic
    server actions alike.

"""

@property
    @classmethod
    @abc.abstractmethod
    def _tag(cls) -> str:
        raise NotImplementedError

@EVENT
class AbstractClientRequest(AbstractServerEvent):
    """Base class for all SSH agent protocol requests.

These are the internal representation the protocol stack parses the
    network input into, to be further handled within the protocol stack.
    Agent implementations normally do not come into contact with these.

"""

_code: ClassVar[int]

@EVENT
class AbstractServerResponse(AbstractServerEvent):
    """Base class for immediate protocol responses."""

_code: ClassVar[int]

@EVENT
class AbstractServerAction(AbstractServerEvent):
    """An automatic action that the protocol stack took on the SSH agent's behalf."""  # noqa: E501

@EVENT
class AbstractProtocolStackFeedback(abc.ABC):
    """Base class for protocol stack feedback to the agent implementation.

Some of the feedback is purely informational.  Other feedback
    constitutes (remote) procedure calls on the agent implementation, in
    the form of [agent actions][AgentAction].

"""

client_initiated: bool

@property
    @classmethod
    @abc.abstractmethod
    def _tag(cls) -> str:
        raise NotImplementedError

@EVENT
class AbstractAgentAction(AbstractProtocolStackFeedback):
    """Base class for actions to be taken by the SSH agent implementation.

These are events the protocol stack expects the SSH agent
    implementation (built on top of the protocol stack) to adequately
    react to, by taking the respective action.

Most actions can only be initiated by the agent client, i.e., they
    are a direct result of the protocol messages received from the
    client (and deserialised and interpreted by the protocol stack).  A
    few select actions are sometimes or always initiated by the protocol
    stack instead, usually because of the stack's state management.

"""

client_initiated: bool

@EVENT
class AbstractAgentEvent(abc.ABC):
    """Base class for events the agent passes to the protocol stack.

This includes [input from the client][ClientInput],
    [responses to agent actions][AgentReaction] and [other input
    from the agent][AgentInput].

"""

@property
    @classmethod
    @abc.abstractmethod
    def _tag(cls) -> str:
        raise NotImplementedError

@EVENT
class AbstractClientInput(AbstractAgentEvent):
    """Base class for client-initiated input to the protocol stack."""

@EVENT
class AbstractAgentReaction(AbstractAgentEvent):
    """Base class for events in response to agent actions.

These communicate the results of [agent actions][AgentAction] which
    the protocol stack instructed the agent implementation to take.

"""

@EVENT
class AbstractAgentInput(AbstractAgentEvent):
    """Base class for agent-initiated input to the protocol stack."""

class SignRequestFlag(enum.IntFlag):
    """Flags for an `SSH_AGENTC_SIGN_REQUEST` request."""

RSA_SHA2_256 = 0x00000002
    RSA_SHA2_512 = 0x00000004

@EVENT
class AbstractKeyConstraint(abc.ABC):
    """Base class for all constraints on keys in the SSH agent protocol."""

_code: ClassVar[int]

@property
    @classmethod
    @abc.abstractmethod
    def _tag(cls) -> str:
        raise NotImplementedError

@EVENT
class TimeoutConstraint(AbstractKeyConstraint):
    """The key is purged from the agent after a certain amount of time."""

_code: ClassVar[int] = 1
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["TimeoutConstraint"]] = "TimeoutConstraint"

deadline: int

@EVENT
class ConfirmUseConstraint(AbstractKeyConstraint):
    """Use of the key must be confirmed by the user."""

_code: ClassVar[int] = 2
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["ConfirmUseConstraint"]] = "ConfirmUseConstraint"

@EVENT
class ExtensionConstraint(AbstractKeyConstraint):
    """A named extension key constraint."""

_code: ClassVar[int] = 255
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["ExtensionConstraint"]] = "ExtensionConstraint"

extension_name: str
    payload: bytes

@EVENT
class BytesFromClient(AbstractClientInput):
    """The client sent some bytes to the agent.

The protocol stack is tasked with parsing and interpreting those
    bytes, if possible.  The bytes need not form a single protocol
    message, or even a complete one.

"""

payload: bytes

@EVENT
class BytesToClient(AbstractAgentAction):
    """The agent shall send some bytes to the client."""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["BytesToClient"]] = "BytesToClient"

payload: bytes

@EVENT
class BaseAddIdentityRequest(AbstractClientRequest):
    """Base class for `SSH_AGENTC_ADD_ID*` requests."""

@EVENT
class AddIdentityUnconstrainedRequest(BaseAddIdentityRequest):
    """An `SSH_AGENTC_ADD_IDENTITY` request."""

_code: ClassVar[int] = _types.SSH_AGENTC.ADD_IDENTITY
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["AddIdentityUnconstrainedRequest"]] = (
        "AddIdentityUnconstrainedRequest"
    )

key_type: str
    key_material: bytes
    comment: str | bytes

@EVENT
class AddIdentityConstrainedRequest(BaseAddIdentityRequest):
    """An `SSH_AGENTC_ADD_ID_CONSTRAINED` request."""

_code: ClassVar[int] = _types.SSH_AGENTC.ADD_ID_CONSTRAINED
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["AddIdentityConstrainedRequest"]] = (
        "AddIdentityConstrainedRequest"
    )

key_type: str
    key_material: bytes
    comment: str | bytes
    constraints: AbstractSet[AbstractKeyConstraint]

@EVENT
class AddSmartcardKeyUnconstrainedRequest(BaseAddIdentityRequest):
    """An `SSH_AGENTC_ADD_SMARTCARD_KEY` request."""

_code: ClassVar[int] = _types.SSH_AGENTC.ADD_SMARTCARD_KEY
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["AddSmartcardKeyUnconstrainedRequest"]] = (
        "AddSmartcardKeyUnconstrainedRequest"
    )

token_id: bytes
    pin: bytes

@EVENT
class AddSmartcardKeyConstrainedRequest(BaseAddIdentityRequest):
    """An `SSH_AGENTC_ADD_SMARTCARD_KEY_CONSTRAINED` request."""

_code: ClassVar[int] = _types.SSH_AGENTC.ADD_SMARTCARD_KEY_CONSTRAINED
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["AddSmartcardKeyConstrainedRequest"]] = (
        "AddSmartcardKeyConstrainedRequest"
    )

token_id: bytes
    pin: bytes
    constraints: AbstractSet[AbstractKeyConstraint]

@EVENT
class BaseRemoveKeyMaterialRequest(AbstractClientRequest):
    """Base class for `SSH_AGENTC_REMOVE_*` requests."""

@EVENT
class RemoveIdentityRequest(BaseRemoveKeyMaterialRequest):
    """An `SSH_AGENTC_REMOVE_IDENTITY` request."""

_code: ClassVar[int] = _types.SSH_AGENTC.REMOVE_IDENTITY
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["RemoveIdentityRequest"]] = "RemoveIdentityRequest"

key_blob: bytes

@EVENT
class RemoveAllIdentitiesRequest(BaseRemoveKeyMaterialRequest):
    """An `SSH_AGENTC_REMOVE_ALL_IDENTITIES` request."""

_code: ClassVar[int] = _types.SSH_AGENTC.REMOVE_ALL_IDENTITIES
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["RemoveAllIdentitiesRequest"]] = (
        "RemoveAllIdentitiesRequest"
    )

@EVENT
class RemoveSmartcardKeyRequest(BaseRemoveKeyMaterialRequest):
    """An `SSH_AGENTC_REMOVE_SMARTCARD_KEY` request."""

_code: ClassVar[int] = _types.SSH_AGENTC.REMOVE_SMARTCARD_KEY
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["RemoveSmartcardKeyRequest"]] = (
        "RemoveSmartcardKeyRequest"
    )

token_id: bytes
    pin: bytes

@EVENT
class ListIdentitiesRequest(AbstractClientRequest):
    """An `SSH_AGENTC_REQUEST_IDENTITIES` request."""

_code: ClassVar[int] = _types.SSH_AGENTC.REQUEST_IDENTITIES
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["ListIdentitiesRequest"]] = "ListIdentitiesRequest"

@EVENT
class SignRequest(AbstractClientRequest):
    """An `SSH_AGENTC_SIGN_REQUEST` request."""

_code: ClassVar[int] = _types.SSH_AGENTC.SIGN_REQUEST
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["SignRequest"]] = "SignRequest"

key_blob: bytes
    payload: bytes
    flags: SignRequestFlag

@EVENT
class LockRequest(AbstractClientRequest):
    """An `SSH_AGENTC_LOCK` request."""

_code: ClassVar[int] = _types.SSH_AGENTC.LOCK
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["LockRequest"]] = "LockRequest"

passphrase: bytes

@EVENT
class UnlockRequest(AbstractClientRequest):
    """An `SSH_AGENTC_UNLOCK` request."""

_code: ClassVar[int] = _types.SSH_AGENTC.UNLOCK
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["UnlockRequest"]] = "UnlockRequest"

passphrase: bytes

@EVENT
class AbstractExtensionRequest(AbstractClientRequest):
    """Base class for known and unknown `SSH_AGENTC_EXTENSION` requests."""

_code: ClassVar[int] = _types.SSH_AGENTC.EXTENSION

@EVENT
class QueryExtensionRequest(AbstractClientRequest):
    """A "query" `SSH_AGENTC_EXTENSION` request."""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["QueryExtensionRequest"]] = "QueryExtensionRequest"

@EVENT
class UnrecognizedExtensionRequest(AbstractExtensionRequest):
    """An `SSH_AGENTC_EXTENSION` request of an unrecognized type."""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["UnrecognizedExtensionRequest"]] = (
        "UnrecognizedExtensionRequest"
    )

extension_type: str
    extension_payload: bytes

@EVENT
class SuccessResponse(AbstractServerResponse):
    """An `SSH_AGENT_SUCCESS` response."""

_code: ClassVar[int] = _types.SSH_AGENT.SUCCESS
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["SuccessResponse"]] = "SuccessResponse"

@EVENT
class FailureResponse(AbstractServerResponse):
    """An `SSH_AGENT_FAILURE` response."""

_code: ClassVar[int] = _types.SSH_AGENT.FAILURE
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["FailureResponse"]] = "FailureResponse"

@EVENT
class SignResponse(AbstractServerResponse):
    """An `SSH_AGENT_SIGN_RESPONSE` response."""

_code: ClassVar[int] = _types.SSH_AGENT.SIGN_RESPONSE
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["SignResponse"]] = "SignResponse"

signature: bytes

@EVENT
class IdentitiesResponse(AbstractServerResponse):
    """An `SSH_AGENT_IDENTITIES_ANSWER` response."""

_code: ClassVar[int] = _types.SSH_AGENT.IDENTITIES_ANSWER
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["IdentitiesResponse"]] = "IdentitiesResponse"

identities: tuple[tuple[bytes, str], ...]

@EVENT
class AbstractExtensionResponse(AbstractServerResponse):
    """Base class for responses to `SSH_AGENTC_EXTENSION` requests."""

@EVENT
class AbstractUnmanagedExtensionResponse(AbstractExtensionResponse):
    """A response to an unmanaged type of `SSH_AGENTC_EXTENSION` request."""

extension_type: str

@EVENT
class UnmanagedExtensionSuccessResponse(AbstractUnmanagedExtensionResponse):
    """A successful response to an `SSH_AGENTC_EXTENSION` request.

May be an `SSH_AGENT_SUCCESS` or an `SSH_AGENT_EXTENSION_RESPONSE`
    response.

Not used for extension requests that the protocol stack knows how
    to handle itself.

"""

_code: ClassVar[
        Literal[_types.SSH_AGENT.SUCCESS, _types.SSH_AGENT.EXTENSION_RESPONSE]
    ]
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["UnmanagedExtensionSuccessResponse"]] = (
        "UnmanagedExtensionSuccessResponse"
    )

response_data: bytes

@EVENT
class UnmanagedExtensionFailureResponse(AbstractUnmanagedExtensionResponse):
    """An `SSH_AGENT_EXTENSION_FAILURE` response.

Not used for extension requests that the protocol stack knows how to
    handle itself.

"""

_code: ClassVar[int] = _types.SSH_AGENT.EXTENSION_FAILURE
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["UnmanagedExtensionFailureResponse"]] = (
        "UnmanagedExtensionFailureResponse"
    )

@EVENT
class QueryExtensionResponse(AbstractExtensionResponse):
    """A successful response to `SSH_AGENTC_EXTENSION` "query" requests."""

_code: ClassVar[
        Literal[_types.SSH_AGENT.SUCCESS, _types.SSH_AGENT.EXTENSION_RESPONSE]
    ]
    # pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["QueryExtensionResponse"]] = (
        "QueryExtensionResponse"
    )

supported_extensions: tuple[str, ...]
    extension_type: str = "query"

@EVENT
class ServerLocked(AbstractServerAction):
    """The protocol stack locked the agent against further private key use."""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["ServerLocked"]] = "ServerLocked"

passphrase: str

@EVENT
class ServerUnlocked(AbstractServerAction):
    """The protocol stack unlocked the agent for further private key use."""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["ServerUnlocked"]] = "ServerUnlocked"

@EVENT
class UnfulfillableRejectedRequest(AbstractServerAction):
    """The request was automatically rejected because it is unfulfillable.

The request requires something that the SSH agent does not have, or
    that does not match the agent's state.

"""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["UnfulfillableRejectedRequest"]] = (
        "UnfulfillableRejectedRequest"
    )

request: AbstractClientRequest

@EVENT
class MissingKeyRejectedRequest(AbstractServerAction):
    """The request was automatically rejected due to missing key material."""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["MissingKeyRejectedRequest"]] = (
        "MissingKeyRejectedRequest"
    )

request: SignRequest | BaseRemoveKeyMaterialRequest

@EVENT
class LockedAgentRejectedRequest(AbstractServerAction):
    """The request was automatically rejected because the agent is locked."""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["LockedAgentRejectedRequest"]] = (
        "LockedAgentRejectedRequest"
    )

@EVENT
class StoreKey(AbstractAgentAction):
    """The agent shall store the given key in its keystore.

The agent implementation shall send a [`KeyAdded`][] event if it
    successfully stored the key material, or a [`UserActionFailed`][]
    event if it could not or refused to store the key material.

This action can only be initiated by the client.

The [`TimeoutConstraint`][] and the [`ConfirmUseConstraint`][], if
    present, are informative; they are enforced by the protocol stack
    automatically.  All other constraints must be handled by the agent
    implementation.

"""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["StoreKey"]] = "StoreKey"

key_type: str
    key_material: bytes
    comment: str | bytes
    constraints: AbstractSet[AbstractKeyConstraint]
    client_initiated: Literal[True] = dataclasses.field(
        default=True, init=False
    )

@EVENT
class LoadSmartcardKeys(AbstractAgentAction):
    """The agent shall store the keys from the named hardware token in its keystore.

The agent implementation shall attempt to locate and access the
    named hardware token, using the given PIN if necessary, and load all
    supported keys from it, with the given constraints.  The agent
    implementation shall send a [`SmartcardKeysAdded`][] event with the
    loaded keys' public keys to the protocol stack if any keys were
    successfully loaded from the hardware token, or
    a [`UserActionFailed`][] event otherwise.

This action can only be initiated by the client.

"""  # noqa: E501

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["LoadSmartcardKeys"]] = "LoadSmartcardKeys"

token_id: str
    pin: str
    constraints: AbstractSet[AbstractKeyConstraint]
    client_initiated: Literal[True] = dataclasses.field(
        default=True, init=False
    )

@EVENT
class PurgeKey(AbstractAgentAction):
    """The agent shall purge the given key from its keystore.

This action may be initiated by the client or by the protocol stack.
    The key may be a smartcard key.

"""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["PurgeKey"]] = "PurgeKey"

public_key: bytes
    token_id: str | None

@EVENT
class PerformSignature(AbstractAgentAction):
    """The agent shall sign the given data with the given key.

This action can only be initiated by the client.

The key constraints list contains only those constraints not already
    handled by the protocol stack.

"""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["PerformSignature"]] = "PerformSignature"

public_key: bytes
    payload: bytes
    signature_flags: SignRequestFlag
    constraints: AbstractSet[AbstractKeyConstraint]
    client_initiated: Literal[True] = dataclasses.field(
        default=True, init=False
    )

@EVENT
class ConfirmKeyUse(AbstractAgentAction):
    """The agent shall prompt the user for confirmation to use the specified key.

The agent implementation shall present an interactive dialog to the
    user, asking for permission to use the specified key.  (For
    compatibility with other implementations, the message should refer
    to the key using its base64-encoded SHA256 fingerprint and its
    comment.)  The agent implementation shall send
    a [`KeyUseConfirmed`][] event if permission was granted, and
    a [`UserActionFailed`][] event if not.

"""  # noqa: E501

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["ConfirmKeyUse"]] = "ConfirmKeyUse"

public_key: bytes
    client_initiated: Literal[False] = dataclasses.field(
        default=False, init=False
    )

@EVENT
class AgentActionFailed(AbstractAgentReaction):
    """The agent implementation failed to fulfill the protocol stack's request."""  # noqa: E501

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["AgentActionFailed"]] = "AgentActionFailed"

@EVENT
class KeyUseConfirmed(AbstractAgentReaction):
    """The agent implementation received permission to use a specific key."""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["KeyUseConfirmed"]] = "KeyUseConfirmed"

public_key: bytes

@EVENT
class KeyAdded(AbstractAgentReaction):
    """The agent implementation added this key."""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["KeyAdded"]] = "KeyAdded"

public_key: bytes
    comment: str | bytes

@EVENT
class SmartcardKeysAdded(AbstractAgentReaction):
    """The agent implementation added these keys from a hardware token."""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["SmartcardKeysAdded"]] = "SmartcardKeysAdded"

token_id: str
    public_keys_and_comments: tuple[tuple[bytes, str | bytes], ...]

@EVENT
class TimeProgressed(AbstractAgentInput):
    """The agent implementation signals the passage of time."""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["TimeProgressed"]] = "TimeProgressed"

timestamp: int | float

@EVENT
class NextTimedEventAt(AbstractProtocolStackFeedback):
    """The protocol stack reports the time of the next timed event."""

# pyrefly: ignore [bad-override]
    _tag: ClassVar[Literal["NextTimedEventAt"]] = "NextTimedEventAt"

timestamp: int | float
    client_initiated: Literal[False] = dataclasses.field(
        default=False, init=False
    )

ServerAction = Union[
    LockedAgentRejectedRequest,
    MissingKeyRejectedRequest,
    UnfulfillableRejectedRequest,
    ServerUnlocked,
    ServerLocked,
]
UnmanagedExtensionResponse = Union[
    UnmanagedExtensionFailureResponse, UnmanagedExtensionSuccessResponse
]
ExtensionResponse = Union[QueryExtensionResponse, UnmanagedExtensionResponse]
ServerResponse = Union[
    ExtensionResponse,
    IdentitiesResponse,
    SignResponse,
    FailureResponse,
    SuccessResponse,
]
ExtensionRequest = Union[UnrecognizedExtensionRequest]
RemoveKeyMaterialRequest = Union[
    RemoveIdentityRequest,
    RemoveAllIdentitiesRequest,
    RemoveSmartcardKeyRequest,
]
AddIdentityRequest = Union[
    AddIdentityUnconstrainedRequest,
    AddIdentityConstrainedRequest,
    AddSmartcardKeyUnconstrainedRequest,
    AddSmartcardKeyConstrainedRequest,
]
ClientRequest = Union[
    QueryExtensionRequest,
    ExtensionRequest,
    UnlockRequest,
    LockRequest,
    SignRequest,
    ListIdentitiesRequest,
    RemoveKeyMaterialRequest,
    AddIdentityRequest,
]
ServerEvent = Union[ServerAction, ServerResponse, ClientRequest]
AgentAction = Union[
    ConfirmKeyUse,
    PerformSignature,
    PurgeKey,
    LoadSmartcardKeys,
    StoreKey,
    BytesToClient,
]
ProtocolStackFeedback = Union[NextTimedEventAt, AgentAction]
AgentInput = Union[TimeProgressed]
AgentReaction = Union[
    SmartcardKeysAdded, KeyAdded, KeyUseConfirmed, AgentActionFailed
]
ClientInput = Union[BytesFromClient]
AgentEvent = Union[AgentInput, AgentReaction, ClientInput]
KeyConstraint = Union[
    ExtensionConstraint, ConfirmUseConstraint, TimeoutConstraint
]

fakesshagent.git

Sketch events for Sans-I/O version