Skip to content

fix: Fix legacy authentication for all PLCs with legacy firmware. #849

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
bvanelli wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

fix: Fix legacy authentication for all PLCs with legacy firmware. #849

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
88 changes: 65 additions & 23 deletions s7commplus/async_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,14 +44,23 @@
_parse_protection_level_response,
_set_s7_groups,
)
from .legitimation import (
build_legacy_response,
build_new_response,
decide_legitimation_mode,
derive_legitimation_key,
extract_session_version_string,
)
from .protocol import (
READ_FUNCTION_CODES,
S7COMMPLUS_LOCAL_TSAP,
S7COMMPLUS_REMOTE_TSAP,
AccessLevel,
DataType,
ElementID,
FunctionCode,
Ids,
LegitimationType,
ObjectId,
Opcode,
ProtocolVersion,
Expand Down Expand Up @@ -226,43 +235,76 @@ async def authenticate(self, password: str, username: str = "") -> None:

Args:
password: PLC password
username: Username for new-style auth (optional)
username: Username for the new-mode exchange (leave empty for legacy)

Raises:
S7ConnectionError: If not connected, TLS not active, or auth fails
S7ConnectionError: If not connected, TLS is not active, the firmware
does not support legitimation, or the password was refused
"""
if not self._connected:
from snap7.error import S7ConnectionError
from snap7.error import S7ConnectionError

if not self._connected:
raise S7ConnectionError("Not connected")

if not self._tls_active or self._oms_secret is None:
from snap7.error import S7ConnectionError

if not self._tls_active:
raise S7ConnectionError("Legitimation requires TLS. Connect with use_tls=True.")

level_before = self._protection_level
if level_before is None:
raise S7ConnectionError("PLC does not report a protection level, so legitimation cannot be verified")

if level_before <= AccessLevel.FULL_ACCESS:
logger.info("PLC already grants full access, legitimation is not required")
return
if not password:
logger.warning(f"PLC restricts access (level {level_before}) but no password was provided")
return

# Step 1: Auto-detect legacy vs new from the firmware version
mode = self._decide_legitimation_mode()
if mode is None:
raise S7ConnectionError("PLC firmware version does not support legitimation")
logger.info(f"Using {mode.name.lower()} legitimation")

# Step 2: Get challenge from PLC via GetVarSubStreamed
challenge = await self._get_legitimation_challenge()
logger.info(f"Received legitimation challenge ({len(challenge)} bytes)")

from .legitimation import build_legacy_response, build_new_response

if username:
response_data = build_new_response(password, challenge, self._oms_secret, username)
await self._send_legitimation_new(response_data)
if mode is LegitimationType.LEGACY:
# A legacy challenge is XORed with a SHA-1 password hash, so it is that long.
if len(challenge) != 20:
raise S7ConnectionError(f"Unexpected legacy challenge length: {len(challenge)}")
await self._send_legitimation_legacy(build_legacy_response(password, challenge))
else:
try:
response_data = build_new_response(password, challenge, self._oms_secret, "")
await self._send_legitimation_new(response_data)
except NotImplementedError:
response_data = build_legacy_response(password, challenge)
await self._send_legitimation_legacy(response_data)

logger.info("PLC legitimation completed successfully")
if self._oms_secret is None:
raise S7ConnectionError(
"New legitimation requires the TLS OMS exporter secret, which could not be derived from this TLS session."
)
await self._send_legitimation_new(build_new_response(password, challenge, self._oms_secret, username))
# The PLC rolls the key after every attempt; mirror it so a second
# legitimation on the same session encrypts with the same key.
self._oms_secret = derive_legitimation_key(self._oms_secret)

# Renew protection level
# Step 3: Renew protection level, which is what verifies the outcome
self._protection_level = await self._get_effective_protection_level()
if self._protection_level is not None:
logger.info(f"PLC reports protection level: {self._protection_level}")
if self._protection_level is None:
raise S7ConnectionError("Legitimation outcome is unverifiable: the PLC stopped reporting its protection level")
if self._protection_level >= level_before:
raise S7ConnectionError(
f"Legitimation failed, protection level unchanged at {self._protection_level}: the password was refused"
)
logger.info(f"PLC legitimation completed, protection level {level_before} -> {self._protection_level}")

def _decide_legitimation_mode(self) -> Optional[LegitimationType]:
"""Return the legitimation exchange the PLC firmware expects, None if unsupported."""
if self._server_session_version is None:
return None
version_string = extract_session_version_string(self._server_session_version)
if version_string is None:
logger.warning("ServerSessionVersion carries no device string, cannot pick a legitimation mode")
return None
logger.debug(f"PLC device string: {version_string}")
return decide_legitimation_mode(version_string)

async def _activate_tls(
self,
Expand Down
105 changes: 74 additions & 31 deletions s7commplus/connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,15 +51,24 @@
from snap7.connection import ISOTCPConnection

from .codec import decode_header, encode_header, encode_object_qualifier, parse_create_object_attributes
from .legitimation import (
build_legacy_response,
build_new_response,
decide_legitimation_mode,
derive_legitimation_key,
extract_session_version_string,
)
from .protocol import (
READ_FUNCTION_CODES,
S7COMMPLUS_LOCAL_TSAP,
S7COMMPLUS_REMOTE_TSAP,
AccessLevel,
DataType,
ElementID,
FunctionCode,
Ids,
LegitimationId,
LegitimationType,
ObjectId,
Opcode,
ProtocolVersion,
Expand Down Expand Up @@ -194,7 +203,10 @@ def _build_set_variable_payload(in_object_id: int, address: int, value: bytes) -


def _check_set_variable_response(payload: bytes) -> None:
"""Raise when a SetVariable response reports a non-zero return value."""
"""Raise when a SetVariable response reports a refused legitimation.

Reference: thomas-v2/S7CommPlusDriver/Legitimation/Legitimation.cs
"""
from snap7.error import S7ConnectionError

if not payload:
Expand All @@ -203,7 +215,9 @@ def _check_set_variable_response(payload: bytes) -> None:
return_value, _ = decode_uint64_vlq(payload, 0)
except ValueError as exc:
raise S7ConnectionError(f"Malformed SetVariable response: {exc}") from exc
if return_value != 0:
# The low 16 bits of the status word are a signed error code; the reference
# driver casts them with (Int16) and rejects negatives, i.e. the sign bit.
if return_value & 0x8000:
raise S7ConnectionError(f"Legitimation rejected by PLC: return_value=0x{return_value:X}")


Expand Down Expand Up @@ -470,7 +484,8 @@ def connect(
self._session_activate()
self._post_auth_legitimation(password=self._connect_password)

# Only a session that completed setup answers attribute reads
# Only a session that completed setup answers attribute reads; the
# V1-initial band falls back to legacy PUT/GET and never gets here.
if self._session_setup_ok:
self._protection_level = self._get_effective_protection_level()
if self._protection_level is not None:
Expand All @@ -497,48 +512,76 @@ def authenticate(self, password: str, username: str = "") -> None:

Args:
password: PLC password
username: Username for new-style auth (optional)
username: Username for the new-mode exchange (leave empty for legacy)

Raises:
S7ConnectionError: If not connected, TLS not active, or auth fails
S7ConnectionError: If not connected, TLS is not active, the firmware
does not support legitimation, or the password was refused
"""
if not self._connected:
from snap7.error import S7ConnectionError
from snap7.error import S7ConnectionError

if not self._connected:
raise S7ConnectionError("Not connected")

if not self._tls_active or self._oms_secret is None:
from snap7.error import S7ConnectionError

if not self._tls_active:
raise S7ConnectionError("Legitimation requires TLS. Connect with use_tls=True.")

# Step 1: Get challenge from PLC via GetVarSubStreamed
level_before = self._protection_level
if level_before is None:
raise S7ConnectionError("PLC does not report a protection level, so legitimation cannot be verified")

if level_before <= AccessLevel.FULL_ACCESS:
logger.info("PLC already grants full access, legitimation is not required")
return
if not password:
logger.warning(f"PLC restricts access (level {level_before}) but no password was provided")
return

# Step 1: Auto-detect legacy vs new from the firmware version
mode = self._decide_legitimation_mode()
if mode is None:
raise S7ConnectionError("PLC firmware version does not support legitimation")
logger.info(f"Using {mode.name.lower()} legitimation")

# Step 2: Get challenge from PLC via GetVarSubStreamed
challenge = self._get_legitimation_challenge()
logger.info(f"Received legitimation challenge ({len(challenge)} bytes)")

# Step 2: Build response (auto-detect legacy vs new)
from .legitimation import build_legacy_response, build_new_response

if username:
# New-style auth with username always uses AES-256-CBC
response_data = build_new_response(password, challenge, self._oms_secret, username)
self._send_legitimation_new(response_data)
if mode is LegitimationType.LEGACY:
# A legacy challenge is XORed with a SHA-1 password hash, so it is that long.
if len(challenge) != 20:
raise S7ConnectionError(f"Unexpected legacy challenge length: {len(challenge)}")
self._send_legitimation_legacy(build_legacy_response(password, challenge))
else:
# Try new-style first, fall back to legacy SHA-1 XOR
try:
response_data = build_new_response(password, challenge, self._oms_secret, "")
self._send_legitimation_new(response_data)
except NotImplementedError:
# cryptography package not available, use legacy
response_data = build_legacy_response(password, challenge)
self._send_legitimation_legacy(response_data)

logger.info("PLC legitimation completed successfully")
if self._oms_secret is None:
raise S7ConnectionError(
"New legitimation requires the TLS OMS exporter secret, which could not be derived from this TLS session."
)
self._send_legitimation_new(build_new_response(password, challenge, self._oms_secret, username))
# The PLC rolls the key after every attempt; mirror it so a second
# legitimation on the same session encrypts with the same key.
self._oms_secret = derive_legitimation_key(self._oms_secret)

# Renew protection level
# Step 3: Renew protection level, which is what verifies the outcome
self._protection_level = self._get_effective_protection_level()
if self._protection_level is not None:
logger.info(f"PLC reports protection level: {self._protection_level}")
if self._protection_level is None:
raise S7ConnectionError("Legitimation outcome is unverifiable: the PLC stopped reporting its protection level")
if self._protection_level >= level_before:
raise S7ConnectionError(
f"Legitimation failed, protection level unchanged at {self._protection_level}: the password was refused"
)
logger.info(f"PLC legitimation completed, protection level {level_before} -> {self._protection_level}")

def _decide_legitimation_mode(self) -> Optional[LegitimationType]:
"""Return the legitimation exchange the PLC firmware expects, None if unsupported."""
if self._server_session_version is None:
return None
version_string = extract_session_version_string(self._server_session_version)
if version_string is None:
logger.warning("ServerSessionVersion carries no device string, cannot pick a legitimation mode")
return None
logger.debug(f"PLC device string: {version_string}")
return decide_legitimation_mode(version_string)

def _get_effective_protection_level(self) -> Optional[int]:
"""Read the session's effective protection level (see `AccessLevel`), None if request failed."""
Expand Down
Loading
Loading