Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions custom_components/lock_code_manager/domain/credentials.py
Original file line number Diff line number Diff line change
Expand Up @@ -269,6 +269,13 @@ class CredentialTypeCapability:
``supports_learn`` is True when the lock can enroll the credential at the
device (for example a fingerprint learn flow) rather than being told the
value.

Length convention shared by every provider: a non-positive ``max_length``
means "no advertised maximum / unknown" -- never a literal zero-length
limit, which would be meaningless -- so providers map an absent or
unreadable maximum to ``0`` (Matter's ``max_pin_length or 0`` idiom). A
non-positive ``min_length`` means "no minimum". ``length_bounds`` applies
this normalization; do not emit a literal ``0`` to express a real limit.
"""

num_slots: int
Expand Down Expand Up @@ -335,6 +342,26 @@ def bounded_slot_count(self, credential_type: CredentialType) -> int | None:
return None
return capability.num_slots

def length_bounds(
self, credential_type: CredentialType
) -> tuple[int, int | None] | None:
"""
Return the effective ``(min, max)`` value length for a credential type.

``None`` when the type is unsupported. A non-positive advertised
bound means "unbounded" rather than a literal limit: Matter reports
``max_pin_length`` as ``... or 0``, where ``0`` is "unknown", so it
normalizes to no upper bound (``max`` of ``None``). A non-positive
minimum normalizes to ``0`` (no minimum).
"""
cap = self.capability_for(credential_type)
if cap is None:
return None
return (
max(cap.min_length, 0),
cap.max_length if cap.max_length > 0 else None,
)


def credential_from_slot(slot: int, state: SlotCredential) -> Credential:
"""
Expand Down
55 changes: 54 additions & 1 deletion custom_components/lock_code_manager/domain/slot_coordinator.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
HomeAssistant,
callback,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers.event import async_track_state_change_event
from homeassistant.helpers.issue_registry import (
IssueSeverity,
Expand All @@ -43,6 +43,7 @@

from ..const import ATTR_IN_SYNC, DOMAIN, EVENT_CREDENTIAL_USED
from .config import EntryConfig
from .credentials import CredentialType
from .names import name_error, normalize_name
from .queries import get_entry_config

Expand Down Expand Up @@ -245,10 +246,22 @@ async def async_request_pin_update(self, value: str) -> None:
Normalizing whitespace and the empty-PIN side effect (disabling
the slot on an active slot whose PIN was cleared) live here so
entities do not have to coordinate sibling state themselves.

A non-empty PIN is validated against every bound lock's advertised
length range before it is written; an empty PIN clears the slot and
is exempt. This is the authoritative gate for BOTH ends: the text
entity keeps ``native_min`` and ``native_max`` permissive so Home
Assistant's ``text.set_value`` service neither rejects the empty clear
nor pre-empts the per-lock error built here -- and so a lock
advertising a limit tighter than it really accepts cannot silently
stop the keystrokes with no message at all.
"""
if not value.strip():
value = ""

if value:
self._validate_credential_length(value, CredentialType.PIN)

updates: dict[str, Any] = {CONF_PIN: value}
if not value and self.is_enabled:
_LOGGER.debug(
Expand All @@ -259,6 +272,46 @@ async def async_request_pin_update(self, value: str) -> None:

self._write_config_fields(updates)

def _validate_credential_length(
self, value: str, credential_type: CredentialType
) -> None:
"""
Reject ``value`` if it violates any bound lock's length range.

Authoritative gate for credential length. Iterates every bound lock so
the error names each offending lock with its required range. The lock
set is the entry-wide ``runtime_data.locks`` -- the same set the text
entity mirrors in ``self.locks`` to size its surfaced bounds, since LCM
binds every lock to every slot; a future per-slot binding must update
both sites together. Locks whose capabilities are not cached
(disconnected or not yet probed) and locks that do not advertise
``credential_type`` are skipped -- the write proceeds rather than
blocking on unknown limits, and the sync layer surfaces any later
device rejection.
"""
length = len(value)
violations: list[str] = []
for lock in self._config_entry.runtime_data.locks.values():
caps = lock.cached_capabilities
if caps is None:
continue
bounds = caps.length_bounds(credential_type)
if bounds is None:
continue
lo, hi = bounds
if length < lo or (hi is not None and length > hi):
required = (
f"at least {lo} characters"
if hi is None
else f"{lo}-{hi} characters"
)
violations.append(f"{required} for {lock.display_name}")
if violations:
raise ServiceValidationError(
f"{credential_type.value.upper()} length {length} is not accepted "
f"by all locks: {'; '.join(violations)}"
)

async def async_request_active_toggle(self, enabled: bool) -> None:
"""
Apply an enabled/disabled toggle requested by the switch entity.
Expand Down
13 changes: 13 additions & 0 deletions custom_components/lock_code_manager/providers/_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -1409,6 +1409,19 @@ async def async_get_usercodes(
"""
return await self._project_users_to_slots(CredentialType.PIN, slots)

@final
@property
def cached_capabilities(self) -> LockCapabilities | None:
"""
Return the already-probed capabilities, or ``None``. Never performs I/O.

Synchronous read of the same cache ``_get_cached_capabilities``
populates. Lets synchronous callers (e.g. the PIN text entity sizing
its length bounds) consult capabilities without awaiting; an unprobed
or disconnected lock reads ``None`` and contributes no constraint.
"""
return self._capabilities_cache

@final
async def _get_cached_capabilities(self) -> LockCapabilities:
"""
Expand Down
37 changes: 35 additions & 2 deletions custom_components/lock_code_manager/text.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,8 +49,10 @@ def add_standard_text_entities(slot_num: int, ent_reg: er.EntityRegistry) -> Non
class LockCodeManagerText(BaseLockCodeManagerEntity, TextEntity):
"""Text entity for lock code manager."""

_attr_native_min = 0
_attr_native_max = 9999
# Defaults for keys with no length constraint (the slot name) and the
# fallback when bound locks advertise nothing or an unsatisfiable range.
_DEFAULT_MIN = 0
_DEFAULT_MAX = 9999

def __init__(
self,
Expand All @@ -67,6 +69,37 @@ def __init__(
)
self._attr_mode = text_mode

@property
def native_min(self) -> int:
"""
Return the minimum value length -- always the permissive default.

The advertised per-lock minimum is deliberately NOT surfaced here.
Home Assistant's ``text.set_value`` service rejects
``len(value) < native_min`` before the value reaches the coordinator,
which would block the empty string that clears a slot and would replace
the coordinator's per-lock error with a generic one. The coordinator
(``SlotEntityCoordinator._validate_credential_length``) is the
authoritative minimum gate; an empty PIN is exempt because it clears
the slot.
"""
return self._DEFAULT_MIN

@property
def native_max(self) -> int:
"""
Return the maximum value length -- always the permissive default.

The advertised maximum is deliberately NOT surfaced here, for the same
reason as the minimum. Home Assistant turns ``native_max`` into the
field's ``maxlength``, so a lock advertising a limit lower than it
really accepts would stop the keystrokes with no message at all: the
field simply refuses to grow and nothing says why. The coordinator
refuses the write instead, naming the lock and the range it claims,
which is what somebody needs to see to recognise a bad advertisement.
"""
return self._DEFAULT_MAX

@property
def native_value(self) -> str | None:
"""Return native value."""
Expand Down
33 changes: 33 additions & 0 deletions tests/providers/test_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,39 @@ def teardown_push_subscription(self) -> None:
self.unsubscribe_calls += 1


class _CapsLock(MockLCMLock):
"""Mock lock that advertises PIN capabilities."""

async def async_get_capabilities(self) -> LockCapabilities:
"""Report a single PIN credential type with a 4-8 length range."""
return LockCapabilities(
supports_user_management=True,
max_users=30,
credential_types={
CredentialType.PIN: CredentialTypeCapability(
num_slots=30, min_length=4, max_length=8, supports_learn=False
)
},
)


async def test_cached_capabilities_exposes_warmed_cache(hass: HomeAssistant):
"""cached_capabilities is None until probed, then returns the cached snapshot."""
entity_reg = er.async_get(hass)
config_entry = MockConfigEntry(domain=DOMAIN)
config_entry.add_to_hass(hass)
lock_entity = entity_reg.async_get_or_create(
"lock", "test", "caps_lock", config_entry=config_entry
)
lock = _CapsLock(hass, dr.async_get(hass), entity_reg, config_entry, lock_entity)

assert lock.cached_capabilities is None

caps = await lock._get_cached_capabilities()
assert lock.cached_capabilities is caps
assert lock.cached_capabilities.length_bounds(CredentialType.PIN) == (4, 8)


async def test_base(hass: HomeAssistant):
"""Test base class."""
entity_reg = er.async_get(hass)
Expand Down
34 changes: 34 additions & 0 deletions tests/test_credentials.py
Original file line number Diff line number Diff line change
Expand Up @@ -270,6 +270,40 @@ def test_credential_types_is_snapshotted(self) -> None:
caps.credential_types[CredentialType.RFID] = pin_cap # type: ignore[index]


def _caps(min_length: int, max_length: int) -> LockCapabilities:
"""Build a single-PIN-type LockCapabilities with the given length bounds."""
return LockCapabilities(
supports_user_management=True,
max_users=30,
credential_types={
CredentialType.PIN: CredentialTypeCapability(
num_slots=30,
min_length=min_length,
max_length=max_length,
supports_learn=False,
)
},
)


class TestLengthBounds:
"""LockCapabilities.length_bounds normalizes per-type length limits."""

def test_returns_min_and_max_for_supported_type(self) -> None:
assert _caps(4, 8).length_bounds(CredentialType.PIN) == (4, 8)

def test_unsupported_type_returns_none(self) -> None:
assert _caps(4, 8).length_bounds(CredentialType.RFID) is None

def test_non_positive_max_means_unbounded(self) -> None:
# Matter reports max_pin_length as `... or 0` -- 0 is "unknown", not
# "zero characters", so it must normalize to no upper bound.
assert _caps(4, 0).length_bounds(CredentialType.PIN) == (4, None)

def test_negative_min_clamps_to_zero(self) -> None:
assert _caps(-1, 8).length_bounds(CredentialType.PIN) == (0, 8)


class TestProjectionHelpers:
"""Pure 1:1:1 projection between a managed slot and the User/Credential model."""

Expand Down
Loading
Loading