Add PC/SC backend adapter (#3)

Author: shinaji

pyproject.toml

@@ -88,7 +88,7 @@ known-third-party = []
 root = ["src"]
 
 [tool.ty.src]
-include = ["src/ntag424dna", "tests"]
+include = ["src/schnee", "tests"]
 
 
 [tool.pytest.ini_options]

src/schnee/__init__.py

@@ -0,0 +1,3 @@
+from .contracts import ProfileBackend, ProfileReaderBackend
+from .core import Backend
+from .pcsc import PcscApduClient, PcscBackend

src/schnee/adapters/__init__.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from schnee.adapters.ntag.profile import ChangePlan, TagProfile
+
+
+@runtime_checkable
+class ProfileReaderBackend(Protocol):
+    """Backend adapter that can read an NTAG profile."""
+
+    def read_profile(self) -> TagProfile:
+        """Read the current tag profile."""
+
+
+@runtime_checkable
+class ProfileBackend(ProfileReaderBackend, Protocol):
+    """Backend adapter that can read and write an NTAG profile."""
+
+    def apply_plan(self, plan: ChangePlan) -> TagProfile:
+        """Apply a profile change plan."""

src/schnee/adapters/backend/__init__.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from .pcsc import PcscBackend
+
+if TYPE_CHECKING:
+    from .contracts import ProfileBackend
+
+
+class Backend:
+    """Backend adapter selector."""
+
+    class BackendError(Exception):
+        """Backend error."""
+
+    class BackendNotFoundError(BackendError):
+        """Backend not found."""
+
+    @classmethod
+    def backend_names(cls) -> list[str]:
+        """Get all backend names."""
+        _ = cls
+        pcsc_names = PcscBackend.pcsc_backend_names()
+        names = [f"pcsc:{name}" for name in pcsc_names]
+        if pcsc_names:
+            names.append("pcsc")
+        return sorted(names)
+
+    @classmethod
+    def get(cls, name: str) -> ProfileBackend:
+        """Get a backend adapter by name."""
+        _ = cls
+        if name == "pcsc":
+            return PcscBackend.create_pcsc_backend()
+        if name.startswith("pcsc:"):
+            return PcscBackend.create_pcsc_backend(name.removeprefix("pcsc:"))
+        raise cls.BackendNotFoundError

src/schnee/adapters/backend/contracts.py

@@ -0,0 +1,9 @@
+from .backend import PcscBackend
+from .client import PcscApduClient
+from .reader import (
+    PcscConnection,
+    PcscReader,
+    PcscReaderProvider,
+    SmartcardPcscConnection,
+    SmartcardPcscReader,
+)

src/schnee/adapters/backend/core.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from schnee.adapters.ntag.profile import ChangePlan, TagInfo, TagProfile
+
+from .client import PcscApduClient
+from .reader import PcscConnection, PcscReader, PcscReaderProvider
+
+if TYPE_CHECKING:
+    from schnee.adapters.ntag.apdu import CommandAPDU, ResponseAPDU
+
+
+class PcscBackend:
+    """Backend adapter that wraps a PC/SC reader."""
+
+    class PcscBackendError(Exception):
+        """PC/SC backend error."""
+
+    class UnsupportedPlanError(PcscBackendError):
+        """Raised when a write plan is not implemented by the PC/SC backend."""
+
+    class UnsupportedProfileReadError(PcscBackendError):
+        """Raised when full profile reads are not implemented."""
+
+    def __init__(self, reader: PcscReader) -> None:
+        self.reader = reader
+        self.client = PcscApduClient(reader)
+
+    @property
+    def reader_name(self) -> str:
+        """Return the wrapped PC/SC reader name."""
+        return self.client.reader_name
+
+    def connect(self) -> PcscConnection:
+        """Connect to the wrapped PC/SC reader."""
+        return self.client.connect()
+
+    def send_apdu(self, apdu: CommandAPDU | list[int]) -> ResponseAPDU:
+        """Transmit an APDU through the wrapped PC/SC reader."""
+        return self.client.send_apdu(apdu)
+
+    def read_profile(self) -> TagProfile:
+        """Read the currently reachable NTAG profile."""
+        msg = "PC/SC full profile reads are not implemented yet"
+        raise self.UnsupportedProfileReadError(msg)
+
+    def read_tag_info(self) -> TagInfo:
+        """Read the currently reachable NTAG tag identity summary."""
+        uid = self._read_uid()
+        return TagInfo(
+            uid=uid,
+            features=["pcsc"],
+        )
+
+    def apply_plan(self, plan: ChangePlan) -> TagProfile:
+        """Apply a profile change plan through PC/SC."""
+        _ = plan
+        msg = "PC/SC profile writes are not implemented yet"
+        raise self.UnsupportedPlanError(msg)
+
+    def _read_uid(self) -> str | None:
+        """Read UID using the common PC/SC contactless reader command."""
+        response = self.send_apdu([0xFF, 0xCA, 0x00, 0x00, 0x00])
+        if not response.ok:
+            return None
+        return bytes(response.data).hex().upper()
+
+    @classmethod
+    def create_pcsc_backend(cls, reader_name: str | None = None) -> PcscBackend:
+        """Create a PC/SC backend adapter."""
+        if reader_name is not None:
+            return cls(reader=PcscReaderProvider.get(reader_name))
+
+        readers = PcscReaderProvider.readers()
+        if not readers:
+            raise PcscReaderProvider.ReaderNotFoundError
+        return cls(reader=readers[0])
+
+    @staticmethod
+    def pcsc_backend_names() -> list[str]:
+        """Return available PC/SC reader backend names without the pcsc prefix."""
+        return list(PcscReaderProvider.reader_names())

src/schnee/adapters/backend/pcsc/__init__.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, ClassVar
+
+from schnee.adapters.ntag.apdu import CommandAPDU, ResponseAPDU
+
+if TYPE_CHECKING:
+    from .reader import PcscConnection, PcscReader
+
+
+class PcscApduClient:
+    """PC/SC APDU transport client."""
+
+    ok_statuses: ClassVar[tuple[tuple[int, int], ...]] = (
+        (0x90, 0x00),
+        (0x91, 0x00),
+        (0x91, 0xAF),
+    )
+
+    class PcscApduClientError(Exception):
+        """PC/SC APDU client error."""
+
+    class ApduStatusError(PcscApduClientError):
+        """Raised when a response APDU has an unsuccessful status."""
+
+        def __init__(self, sw1: int, sw2: int) -> None:
+            self.sw1 = sw1
+            self.sw2 = sw2
+            super().__init__(f"SW1: {hex(sw1)} SW2: {hex(sw2)}")
+
+    def __init__(self, reader: PcscReader) -> None:
+        self.reader = reader
+        self.connection: PcscConnection | None = None
+
+    @property
+    def reader_name(self) -> str:
+        """Return the wrapped PC/SC reader name."""
+        return self.reader.name
+
+    def connect(self) -> PcscConnection:
+        """Connect to the wrapped PC/SC reader."""
+        if self.connection is None:
+            self.connection = self.reader.create_connection()
+            self.connection.connect()
+        return self.connection
+
+    def send_apdu(self, apdu: CommandAPDU | list[int]) -> ResponseAPDU:
+        """Transmit an APDU and return the full response."""
+        command = apdu.to_list() if isinstance(apdu, CommandAPDU) else apdu
+        response, sw1, sw2 = self.connect().transmit(command)
+        return ResponseAPDU(data=response, sw1=sw1, sw2=sw2)
+
+    def send_checked(
+        self,
+        apdu: CommandAPDU | list[int],
+        *,
+        ok_statuses: tuple[tuple[int, int], ...] | None = None,
+    ) -> list[int]:
+        """Transmit an APDU and return data for successful status words."""
+        response = self.send_apdu(apdu)
+        statuses = self.ok_statuses if ok_statuses is None else ok_statuses
+        if (response.sw1, response.sw2) in statuses:
+            return response.data
+        raise self.ApduStatusError(response.sw1, response.sw2)

src/schnee/adapters/backend/pcsc/backend.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Protocol
+
+from smartcard.pcsc.PCSCReader import PCSCReader
+from smartcard.System import readers as smartcard_readers
+
+if TYPE_CHECKING:
+    from smartcard.CardConnection import CardConnection
+
+
+class PcscConnection(Protocol):
+    """PC/SC connection interface used by the backend."""
+
+    def connect(self) -> None:
+        """Connect to the card."""
+
+    def transmit(self, command: list[int]) -> tuple[list[int], int, int]:
+        """Transmit a command APDU."""
+
+
+class PcscReader(Protocol):
+    """PC/SC reader interface used by the backend."""
+
+    name: str
+
+    def create_connection(self) -> PcscConnection:
+        """Create a PC/SC connection."""
+
+
+class SmartcardPcscConnection:
+    """Adapter for pyscard connection objects."""
+
+    def __init__(self, connection: CardConnection) -> None:
+        self._connection = connection
+
+    def connect(self) -> None:
+        """Connect to the card."""
+        self._connection.connect()
+
+    def transmit(self, command: list[int]) -> tuple[list[int], int, int]:
+        """Transmit a command APDU."""
+        response, sw1, sw2 = self._connection.transmit(command)
+        return list(response), sw1, sw2
+
+
+class SmartcardPcscReader:
+    """Adapter for pyscard PC/SC reader objects."""
+
+    def __init__(self, reader: PCSCReader) -> None:
+        self._reader = reader
+
+    @property
+    def name(self) -> str:
+        """Return the PC/SC reader name."""
+        return self._reader.name
+
+    def create_connection(self) -> SmartcardPcscConnection:
+        """Create a wrapped PC/SC connection."""
+        return SmartcardPcscConnection(self._reader.createConnection())
+
+
+class PcscReaderProvider:
+    """Discover and wrap pyscard PC/SC readers."""
+
+    class PcscReaderProviderError(Exception):
+        """PC/SC reader provider error."""
+
+    class ReaderNotFoundError(PcscReaderProviderError):
+        """Raised when a requested PC/SC reader is not found."""
+
+    @staticmethod
+    def readers() -> list[PcscReader]:
+        """Return available PC/SC readers."""
+        return [
+            SmartcardPcscReader(reader)
+            for reader in smartcard_readers()
+            if isinstance(reader, PCSCReader)
+        ]
+
+    @classmethod
+    def reader_names(cls) -> list[str]:
+        """Return available PC/SC reader names."""
+        return [reader.name for reader in cls.readers()]
+
+    @classmethod
+    def get(cls, name: str) -> PcscReader:
+        """Return a named PC/SC reader."""
+        for reader in cls.readers():
+            if reader.name == name:
+                return reader
+        raise cls.ReaderNotFoundError