Add MMS ObtainFile service implementation

MatrixEditor · MatrixEditor · commit 8c1774bd183d · 2025-09-28T08:52:06.000+02:00
---

+ Add "put" command to mmsclient.py
+ Add origon configuration control to iedctrl.py
diff --git a/README.md b/README.md
@@ -56,8 +56,7 @@ critical.
 If you decide to build on this work, we encourage you to adopt secure coding
 practices, apply a structured security development process, and consider how
 you’ll generate and track indicators of compromise in line with your specific
-goals. Together, we can strengthen the security and resilience of ICS
-technologies while keeping experimentation safe and responsible.
+goals.
 
 ## License
 
diff --git a/src/icspacket/examples/iedctrl.py b/src/icspacket/examples/iedctrl.py
@@ -54,6 +54,19 @@ def cli2data(value: str) -> Any | None:
     return doc
 
 
+def parse_origin(origin: str) -> tuple[int, bytes]:
+    if ":" not in origin:
+        logging.warning(f"Invalid origin format: {origin!r}")
+        return 0, bytes()
+
+    cat, id = origin.split(":")
+    try:
+        ident = bytes.fromhex(id)
+    except ValueError:
+        ident = id.encode()
+    return int(cat), ident
+
+
 def cli_main():
     from icspacket import __version__
 
@@ -78,6 +91,13 @@ def cli_main():
     )
     group.add_argument("--toggle", action="store_true", help="Toggle the value of the specified variable (queries first and assumes boolean).")
 
+    group = parser.add_argument_group("Control Options")
+    group.add_argument("--origin", metavar="CAT:ID", default=None, help=(
+        "The origin of the control request. \n"
+        "If not specified the default origin will be used."
+    ))
+    group.add_argument("--synchro-check", action="store_true", help="Enables synchro check for the control request.")
+    group.add_argument("--interlock-check", action="store_true", help="Enables interlock check for the control request.")
     # fmt: on
     add_mms_connection_options(parser)
     add_logging_options(parser)
@@ -140,6 +160,13 @@ def cli_main():
         )
         sys.exit(1)
 
+    if args.synchro_check:
+        co.synchro_check = True
+    if args.interlock_check:
+        co.interlock_check = True
+    if args.origin:
+        co.origin_cat, co.origin_ident = parse_origin(args.origin)
+
     if args.toggle:
         logging.debug("Toggle: Requesting current value for node...")
         value = client.get_data_values(ref / "Oper" / "ctlVal")
@@ -196,10 +223,10 @@ def cli_main():
                     _ = client.await_command_termination()
     except LastApplError as e:
         logging.error(
-            "Failed to operate on node %s: error: %s, cause: %s",
+            "Failed to operate on node %s: [bold red]%s[/], cause: [red]%s",
             parts[-1],
-            repr(e.error),
-            repr(e.cause),
+            e.error.name,
+            e.cause.name,
         )
     except ConnectionError as e:
         logging.error(
diff --git a/src/icspacket/examples/mms_utility.py b/src/icspacket/examples/mms_utility.py
@@ -73,7 +73,7 @@ def data_to_str(data: Data) -> str | dict | list:
         case Data.PRESENT.PR_bit_string | Data.PRESENT.PR_booleanArray:
             return data.bit_string.value.to01(sep=" ") if data.bit_string else "<EMPTY>"
         case Data.PRESENT.PR_boolean:
-            return "[green]TRUE[/]" if data.boolean else "FALSE"
+            return "[green]True[/]" if data.boolean else "[red]False[/]"
         case Data.PRESENT.PR_array:
             return [data_to_str(item) for item in data.array]
         case Data.PRESENT.PR_utc_time:
diff --git a/src/icspacket/examples/mmsclient.py b/src/icspacket/examples/mmsclient.py
@@ -14,6 +14,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 # pyright: reportUnusedCallResult=false, reportGeneralTypeIssues=false
+import datetime
 import logging
 import sys
 import pathlib
@@ -29,7 +30,6 @@
 
 from icspacket.core.connection import ConnectionClosedError
 from icspacket.core import hexdump
-from icspacket.proto.cotp.structs import TPDU_Size
 from icspacket.proto.mms._mms import FileName, ServiceError
 from icspacket.proto.mms.connection import MMS_Connection
 from icspacket.proto.mms.exceptions import MMSConnectionError, MMSUnknownServiceError
@@ -202,8 +202,16 @@ def do_ls(self, args) -> None:
                     raw_path = "".join(list(entry.fileName))
                     path = pathlib.Path(raw_path.removeprefix(str(remote_dir) + "/"))
                     # TODO: parse time
+                    try:
+                        mtime = str(
+                            datetime.datetime.strptime(
+                                entry.fileAttributes.lastModified, "%Y%m%d%H%M%S.%fZ"
+                            )
+                        )
+                    except ValueError as e:
+                        mtime = "N/A"
                     table.add_row(
-                        str(path), str(entry.fileAttributes.sizeOfFile), "N/A"
+                        str(path), str(entry.fileAttributes.sizeOfFile), mtime
                     )
                 console.print(table)
             else:
@@ -310,6 +318,29 @@ def do_del(self, args) -> None:
                 error, service_error, f"Could not delete file {str(remote_file_path)!r}"
             )
 
+    put_parser = cmd2.Cmd2ArgumentParser(add_help=True)
+    put_parser.add_argument("local_name", type=str, help="Local file name")
+    put_parser.add_argument(
+        "remote_name", type=str, nargs="?", help="Optional remote file name"
+    )
+
+    @cmd2.with_argparser(put_parser)
+    def do_put(self, args) -> None:
+        """Upload a file to the remote MMS server."""
+        local_file_path = pathlib.Path(args.local_name)
+        if not args.remote_name:
+            args.remote_name = local_file_path.name
+
+        logging.info("Uploading '%s' to '%s'", local_file_path, args.remote_name)
+        try:
+            with local_file_path.open("rb") as local_fp:
+                self.conn.file_transfer(local_file_path, args.remote_name)
+        except MMSConnectionError as error:
+            service_error = error.error
+            self._handle_file_service_error(
+                error, service_error, f"Could not upload file {str(local_file_path)!r}"
+            )
+
     @override
     def pexcept(self, msg: Any, *, end: str = "\n", apply_style: bool = True) -> None:
         match msg:
@@ -328,16 +359,53 @@ def cli_main():
     from icspacket import __version__
     from icspacket.core import logger
 
+    class _HelpAction(argparse.Action):
+        def __init__(
+            self,
+            option_strings,
+            dest=argparse.SUPPRESS,
+            default=argparse.SUPPRESS,
+            help=None,
+        ):
+            super(_HelpAction, self).__init__(
+                option_strings=option_strings,
+                dest=dest,
+                default=default,
+                nargs="?",
+                help=help,
+            )
+
+        def __call__(self, parser, namespace, values, option_string=None):
+            if not values:
+                parser.print_help()
+            else:
+                parser_name = f"{values}_parser"
+                if hasattr(MMSClient, parser_name):
+                    getattr(MMSClient, parser_name).print_help()
+                else:
+                    parser.error(f"no such command: {values}")
+            parser.exit()
+
     parser = argparse.ArgumentParser(
         usage="%(prog)s [options] host [command [args...]]",
+        add_help=False,
     )
+    # parser.register("action", "cmd2_help", Cmd2Action)
     parser.add_argument(
         "-i",
         "--interactive",
         action="store_true",
         help="Continue in interactive mode acter executing the first command (only if given)",
         default=False,
     )
+    parser.add_argument(
+        "-h",
+        "--help",
+        action=_HelpAction,
+        help="Show this help message and exit. Optionally: show help for command",
+        default=None,
+        dest="help",
+    )
     add_mms_connection_options(parser)
     add_logging_options(parser)
 
diff --git a/src/icspacket/proto/mms/connection.py b/src/icspacket/proto/mms/connection.py
@@ -14,15 +14,25 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+import datetime
 import logging
 
 from collections.abc import Iterable
+import os
+from pathlib import Path
+import random
 from typing_extensions import Callable, override
 
 from icspacket.core.connection import connection
 from icspacket.proto.mms._mms import (
+    ApplicationReference,
+    Confirmed_ResponsePDU,
+    FileClose_Response,
+    FileOpen_Response,
+    FileRead_Response,
     GetNamedVariableListAttributes_Request,
     GetNamedVariableListAttributes_Response,
+    ObtainFile_Request,
     Unconfirmed_PDU,
     UnconfirmedService,
 )
@@ -933,6 +943,166 @@ def variable_list_attributes(
         response = self.service_request(service)
         return response.getNamedVariableListAttributes
 
+    # ---------------------------------------------------------------------------
+    # Annex C - File Access service
+    # ---------------------------------------------------------------------------
+    def obtain_file(
+        self,
+        source: str | Path,
+        destination: str | Path,
+        remote: ApplicationReference | None = None,
+    ) -> None:
+        """
+        Perform the MMS :term:`ObtainFile` service to transfer a file
+        between the client and a remote MMS server.
+
+        The ObtainFile service may be used by an MMS client to instruct an
+        MMS server to obtain a specified file from a file server. Depending
+        on the usage, this can either be a request to *pull* a file from
+        a remote source or to *serve* a local file to the requesting MMS
+        peer.
+
+        The method implements the complete reques-response sequence defined
+        in IEC 61850 Annex C, including ``FileOpen``, ``FileRead``, ``FileClose``,
+        and the final ``ObtainFile`` confirmation.
+
+        :param source:
+            Path to the source file. If ``remote`` is provided, this denotes the
+            file path to be retrieved from the remote server. Otherwise, this
+            must point to a local file which will be transferred.
+        :type source: str | Path
+
+        :param destination:
+            Path where the file should be stored on the remote server.
+            For local serving, this is the name advertised to the requesting peer.
+        :type destination: str | Path
+
+        :param remote:
+            Optional reference to an external application server from which
+            the file should be obtained. If ``None``, this MMS connection
+            instance will act as the file source.
+        :type remote: ApplicationReference | None
+
+        :raises MMSServiceError:
+            If the MMS server responds with an unexpected PDU type or an error
+            occurs during the transfer sequence.
+
+        :raises OSError:
+            If the local file system access fails (e.g., file not found or
+            permission denied).
+
+        .. versionadded:: 0.2.4
+        """
+        request = ObtainFile_Request(
+            sourceFile=FileName([os.path.basename(str(source))]),
+            destinationFile=FileName([str(destination)]),
+        )
+        if remote:
+            request.sourceFileServer = remote
+
+        obtain_service = ConfirmedServiceRequest(obtainFile=request)
+        pdu = Confirmed_RequestPDU()
+        pdu.invokeID = self.next_invoke_id
+        pdu.service = obtain_service
+
+        self.send_mms_data(MMSpdu(confirmed_RequestPDU=pdu))
+        response = self.recv_mms_data()
+        error = self._error_from_service_response(
+            obtain_service, response, need_response=remote is not None
+        )
+        if error is not None:
+            raise error
+
+        if remote:
+            return
+
+        if response.present != MMSpdu.PRESENT.PR_confirmed_RequestPDU:
+            raise MMSServiceError(
+                f"Failed ObtainFile request with unexpected response: {response.present!r} - Expected FileOpen"
+            )
+
+        service = response.confirmed_RequestPDU.service
+        if service.present != ConfirmedServiceRequest.PRESENT.PR_fileOpen:
+            raise MMSServiceError(
+                f"Failed ObtainFile request with unexpected response: {service.present!r} - Expected FileOpen"
+            )
+
+        invoke_id = response.confirmed_RequestPDU.invokeID.value
+        # We simply generate a new handle
+        handle = random.randint(0, 0xFFFF)
+        pdu = ConfirmedServiceResponse()
+
+        source_path = Path(source)
+        source_stat = source_path.stat()
+        mtime = datetime.datetime.fromtimestamp(source_stat.st_mtime)
+
+        file_open = FileOpen_Response(frsmID=handle)
+        file_open.fileAttributes.sizeOfFile = source_stat.st_size
+        file_open.fileAttributes.lastModified = mtime.strftime("%Y%m%d%H%M%S.%fZ")
+        pdu.fileOpen = file_open
+
+        response = Confirmed_ResponsePDU()
+        response.invokeID = invoke_id
+        response.service = pdu
+        self.send_mms_data(MMSpdu(confirmed_ResponsePDU=response))
+
+        read_req = self.recv_mms_data()
+        if read_req.present != MMSpdu.PRESENT.PR_confirmed_RequestPDU:
+            raise MMSServiceError(
+                f"Failed ObtainFile request with unexpected response: {read_req.present!r} - Expected FileRead"
+            )
+
+        request = read_req.confirmed_RequestPDU
+        if request.service.present != ConfirmedServiceRequest.PRESENT.PR_fileRead:
+            raise MMSServiceError(
+                f"Failed ObtainFile request with unexpected response: {request.service.present!r} - Expected FileRead"
+            )
+
+        invoke_id = request.invokeID.value
+        pdu = ConfirmedServiceResponse()
+
+        file_read = FileRead_Response()
+        file_read.fileData = source_path.read_bytes()
+        file_read.moreFollows = False
+
+        pdu.fileRead = file_read
+        response = Confirmed_ResponsePDU()
+        response.invokeID = invoke_id
+        response.service = pdu
+        self.send_mms_data(MMSpdu(confirmed_ResponsePDU=response))
+
+        # expect fileClose and obtainFile response
+        close_req = self.recv_mms_data()
+        if close_req.present != MMSpdu.PRESENT.PR_confirmed_RequestPDU:
+            raise MMSServiceError(
+                f"Failed ObtainFile request with unexpected response: {close_req.present!r} - Expected FileClose"
+            )
+
+        close_req = close_req.confirmed_RequestPDU
+        if close_req.service.present != ConfirmedServiceRequest.PRESENT.PR_fileClose:
+            raise MMSServiceError(
+                f"Failed ObtainFile request with unexpected response: {close_req.service.present!r} - Expected FileClose"
+            )
+
+        invoke_id = close_req.invokeID.value
+        pdu = ConfirmedServiceResponse()
+        pdu.fileClose = FileClose_Response(value=None)
+        response = Confirmed_ResponsePDU()
+        response.invokeID = invoke_id
+        response.service = pdu
+        self.send_mms_data(MMSpdu(confirmed_ResponsePDU=response))
+
+        obtain_response = self.recv_mms_data()
+        error = self._error_from_service_response(obtain_service, obtain_response, True)
+        if error is not None:
+            raise error
+
+    # alias to match file_XXX naming convention
+    file_transfer = obtain_file
+    """
+    .. versionadded:: 0.2.4
+    """
+
     # ---------------------------------------------------------------------------
     # Annex D - File Management
     # ---------------------------------------------------------------------------
