Skip to content

iSCSI

iSCSI transports SCSI commands over TCP/IP, allowing remote block storage to appear as local SCSI devices. The initiator side uses iscsiadm to discover targets, log in to sessions, and manage node records (a "node" in open-iscsi is a target+portal+interface tuple).

Device Management

sts.iscsi.device

iSCSI device management.

IscsiDevice pydantic-model

Bases: StorageDevice, NetworkDevice

iSCSI block device accessed over a network connection.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "iSCSI block device accessed over a network connection.",
  "properties": {
    "path": {
      "format": "path",
      "title": "Path",
      "type": "string"
    },
    "name": {
      "title": "Name",
      "type": "string"
    },
    "ip": {
      "title": "Ip",
      "type": "string"
    },
    "port": {
      "default": 3260,
      "maximum": 65535,
      "minimum": 1,
      "title": "Port",
      "type": "integer"
    },
    "size": {
      "title": "Size",
      "type": "integer"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "target_iqn": {
      "title": "Target Iqn",
      "type": "string"
    },
    "initiator_iqn": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Initiator Iqn"
    }
  },
  "required": [
    "path",
    "name",
    "ip",
    "size",
    "target_iqn"
  ],
  "title": "IscsiDevice",
  "type": "object"
}

Fields:

  • model (str | None)
  • name (str)
  • path (Path)
  • size (int)
  • ip (str)
  • target_iqn (str)
  • port (int)
  • initiator_iqn (str | None)

Validators:

  • _derive_fields
Source code in sts_libs/src/sts/iscsi/device.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class IscsiDevice(StorageDevice, NetworkDevice):
    """iSCSI block device accessed over a network connection."""

    name: str  # type: ignore[assignment]
    path: Path  # type: ignore[assignment]
    size: int  # type: ignore[assignment]
    ip: str  # type: ignore[assignment]

    target_iqn: str
    port: int = Field(default=3260, ge=1, le=65535)  # type: ignore[assignment]
    initiator_iqn: str | None = None

    @computed_field  # type: ignore[prop-decorator]
    @property
    def portal(self) -> str:
        """Portal address in IP:Port format."""
        return f'{self.ip}:{self.port}'

    ISCSI_PATH: ClassVar[Path] = Path('/sys/class/iscsi_session')
    DATABASE_PATH: ClassVar[Path] = Path('/var/lib/iscsi')

    @classmethod
    def _parse_device_info(cls, info: dict[str, str]) -> IscsiDevice | None:
        """Parse device info dictionary into an IscsiDevice instance."""
        try:
            return cls(
                name=info['name'],
                path=Path(f'/dev/{info["name"]}'),
                size=int(info['size']),
                ip=info['ip'],
                port=int(info['port']),
                target_iqn=info['target_iqn'],
            )
        except (KeyError, ValueError, DeviceError):
            logger.warning('Failed to create device')
            return None

    @classmethod
    def get_all(cls) -> list[IscsiDevice]:
        """Get all iSCSI devices from active sessions."""
        devices: list[IscsiDevice] = []
        for session in IscsiSession.get_all():
            # Get session details
            session_data = session.get_data_p2()
            if not session_data:
                continue

            # Get portal info (IP:Port)
            portal = session_data.get('Current Portal', '').split(',')[0]
            if not portal:
                continue
            ip, port = portal.split(':')

            # Get disks associated with session
            for disk in session.get_disks():
                if not disk.is_running():
                    continue

                # Get disk size using blockdev
                try:
                    result = run(f'blockdev --getsize64 /dev/{disk.name}')
                    if result.failed:
                        continue
                    size = int(result.stdout)
                except (ValueError, DeviceError):
                    continue

                # Create device object
                device = cls(
                    name=disk.name,
                    path=Path(f'/dev/{disk.name}'),
                    size=size,
                    ip=ip,
                    port=int(port),
                    target_iqn=session.target_iqn,
                )
                devices.append(device)

        return devices

    @classmethod
    def discover(cls, ip: str, port: int = 3260) -> list[str]:
        """Discover available target IQNs via SendTargets discovery."""
        iscsiadm = IscsiAdm()
        result = iscsiadm.discovery(portal=f'{ip}:{port}')
        if result.failed:
            logger.warning('No targets found')
            return []

        targets: list[str] = []
        for line in result.stdout.splitlines():
            # Parse line like: 192.168.1.100:3260,1 iqn.2003-01.target
            parts = line.split()
            if len(parts) > 1 and parts[-1].startswith('iqn.'):
                targets.append(parts[-1])

        return targets

portal property

Portal address in IP:Port format.

discover(ip, port=3260) classmethod

Discover available target IQNs via SendTargets discovery.

Source code in sts_libs/src/sts/iscsi/device.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@classmethod
def discover(cls, ip: str, port: int = 3260) -> list[str]:
    """Discover available target IQNs via SendTargets discovery."""
    iscsiadm = IscsiAdm()
    result = iscsiadm.discovery(portal=f'{ip}:{port}')
    if result.failed:
        logger.warning('No targets found')
        return []

    targets: list[str] = []
    for line in result.stdout.splitlines():
        # Parse line like: 192.168.1.100:3260,1 iqn.2003-01.target
        parts = line.split()
        if len(parts) > 1 and parts[-1].startswith('iqn.'):
            targets.append(parts[-1])

    return targets

get_all() classmethod

Get all iSCSI devices from active sessions.

Source code in sts_libs/src/sts/iscsi/device.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@classmethod
def get_all(cls) -> list[IscsiDevice]:
    """Get all iSCSI devices from active sessions."""
    devices: list[IscsiDevice] = []
    for session in IscsiSession.get_all():
        # Get session details
        session_data = session.get_data_p2()
        if not session_data:
            continue

        # Get portal info (IP:Port)
        portal = session_data.get('Current Portal', '').split(',')[0]
        if not portal:
            continue
        ip, port = portal.split(':')

        # Get disks associated with session
        for disk in session.get_disks():
            if not disk.is_running():
                continue

            # Get disk size using blockdev
            try:
                result = run(f'blockdev --getsize64 /dev/{disk.name}')
                if result.failed:
                    continue
                size = int(result.stdout)
            except (ValueError, DeviceError):
                continue

            # Create device object
            device = cls(
                name=disk.name,
                path=Path(f'/dev/{disk.name}'),
                size=size,
                ip=ip,
                port=int(port),
                target_iqn=session.target_iqn,
            )
            devices.append(device)

    return devices

Session Management

sts.iscsi.session

iSCSI session management.

IscsiSession pydantic-model

Bases: StsBaseModel

iSCSI session representation.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "iSCSI session representation.",
  "properties": {
    "session_id": {
      "minLength": 1,
      "title": "Session Id",
      "type": "string"
    },
    "target_iqn": {
      "minLength": 1,
      "title": "Target Iqn",
      "type": "string"
    },
    "portal": {
      "title": "Portal",
      "type": "string"
    }
  },
  "required": [
    "session_id",
    "target_iqn",
    "portal"
  ],
  "title": "IscsiSession",
  "type": "object"
}

Fields:

  • session_id (str)
  • target_iqn (str)
  • portal (str)
Source code in sts_libs/src/sts/iscsi/session.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class IscsiSession(StsBaseModel):
    """iSCSI session representation."""

    session_id: str = Field(min_length=1)
    target_iqn: str = Field(min_length=1)
    portal: str

    def logout(self) -> bool:
        """Log out from session."""
        result = IscsiAdm().session_logout(self.session_id)
        if result.failed:
            logger.error('Logout failed')
            return False
        return True

    def get_data(self) -> dict[str, str]:
        """Get session data from `iscsiadm -m session -r <sid> -S`.

        Returns a plain dict rather than a typed model on purpose: the set of
        keys varies with the negotiated iSCSI session/connection parameters
        (see `PARAM_MAP` in `.parameters`), so a fixed model would need every
        key declared optional — no safer than a dict, and it would make
        callers that iterate all keys (e.g. `get_parameters()`) more awkward.
        This mirrors the NVMe JSON methods, which keep the same raw-dict
        exception for the same reason.
        """
        result = IscsiAdm().session(**{'-r': self.session_id, '-S': None})
        if result.failed:
            return {}

        data: dict[str, str] = {}
        for line in result.stdout.splitlines():
            if line and not line.startswith('#'):
                key_val = line.split(' = ', 1)
                if len(key_val) == 2:
                    data[key_val[0]] = key_val[1]
        return data

    def get_data_p2(self) -> dict[str, str]:
        """Get session data with print level 2."""
        result = IscsiAdm().session(**{'-r': self.session_id, '-S': None, '-P': '2'})
        if result.failed:
            return {}

        data: dict[str, str] = {}
        for line in result.stdout.splitlines():
            if line and ': ' in line:
                key_val = line.replace('\t', '').split(': ', 1)
                if len(key_val) == 2:
                    data[key_val[0]] = key_val[1]
        return data

    def get_disks(self) -> list[SessionDisk]:
        """Get list of disks attached to session."""
        result = IscsiAdm().session(**{'-r': self.session_id, '-P': '3'})
        if result.failed or 'Attached scsi disk' not in result.stdout:
            return []

        disks: list[SessionDisk] = []
        scsi_pattern = r'scsi(\d+)\s+Channel\s+(\d+)\s+Id\s+(\d+)\s+Lun:\s+(\d+)'
        disk_pattern = r'Attached\s+scsi\s+disk\s+(\w+)\s+State:\s+(\w+)'

        lines = result.stdout.splitlines()
        for i, line in enumerate(lines):
            scsi_match = re.search(scsi_pattern, line)
            if not scsi_match or i + 1 >= len(lines):
                continue

            disk_match = re.search(disk_pattern, lines[i + 1])
            if not disk_match:
                continue

            disks.append(
                SessionDisk(
                    name=disk_match.group(1),
                    state=disk_match.group(2),
                    scsi_n=scsi_match.group(1),
                    channel=scsi_match.group(2),
                    id=scsi_match.group(3),
                    lun=scsi_match.group(4),
                ),
            )

        return disks

    @classmethod
    def get_all(cls) -> list[IscsiSession]:
        """Get list of all iSCSI sessions."""
        result = IscsiAdm().session()
        if result.failed:
            return []

        sessions: list[IscsiSession] = []
        for line in result.stdout.splitlines():
            if not line:
                continue
            parts = line.split()
            if len(parts) < 4:
                continue
            session_id = parts[1].strip('[]')
            portal = parts[2].split(',')[0]
            target_iqn = parts[3]
            sessions.append(cls(session_id=session_id, target_iqn=target_iqn, portal=portal))

        return sessions

    @classmethod
    def get_by_target(cls, target_iqn: str) -> list[IscsiSession]:
        """Get sessions matching target IQN."""
        return [s for s in cls.get_all() if s.target_iqn == target_iqn]

    @classmethod
    def get_by_portal(cls, portal: str) -> list[IscsiSession]:
        """Get sessions matching portal address."""
        return [s for s in cls.get_all() if s.portal == portal]

    def get_parameters(self) -> dict[str, str]:
        """Get negotiated parameters from session."""
        data = self.get_data_p2()
        if not data:
            logger.warning('Failed to get session data')
            return {}

        negotiated: dict[str, str] = {}
        for param_name in PARAM_MAP:
            if param_name not in data:
                logger.warning(f'Parameter {param_name} not found in session data')
                continue
            negotiated[param_name] = data[param_name]

        return negotiated

get_all() classmethod

Get list of all iSCSI sessions.

Source code in sts_libs/src/sts/iscsi/session.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@classmethod
def get_all(cls) -> list[IscsiSession]:
    """Get list of all iSCSI sessions."""
    result = IscsiAdm().session()
    if result.failed:
        return []

    sessions: list[IscsiSession] = []
    for line in result.stdout.splitlines():
        if not line:
            continue
        parts = line.split()
        if len(parts) < 4:
            continue
        session_id = parts[1].strip('[]')
        portal = parts[2].split(',')[0]
        target_iqn = parts[3]
        sessions.append(cls(session_id=session_id, target_iqn=target_iqn, portal=portal))

    return sessions

get_by_portal(portal) classmethod

Get sessions matching portal address.

Source code in sts_libs/src/sts/iscsi/session.py 
146
147
148
149
@classmethod
def get_by_portal(cls, portal: str) -> list[IscsiSession]:
    """Get sessions matching portal address."""
    return [s for s in cls.get_all() if s.portal == portal]

get_by_target(target_iqn) classmethod

Get sessions matching target IQN.

Source code in sts_libs/src/sts/iscsi/session.py 
141
142
143
144
@classmethod
def get_by_target(cls, target_iqn: str) -> list[IscsiSession]:
    """Get sessions matching target IQN."""
    return [s for s in cls.get_all() if s.target_iqn == target_iqn]

get_data()

Get session data from iscsiadm -m session -r <sid> -S.

Returns a plain dict rather than a typed model on purpose: the set of keys varies with the negotiated iSCSI session/connection parameters (see PARAM_MAP in .parameters), so a fixed model would need every key declared optional — no safer than a dict, and it would make callers that iterate all keys (e.g. get_parameters()) more awkward. This mirrors the NVMe JSON methods, which keep the same raw-dict exception for the same reason.

Source code in sts_libs/src/sts/iscsi/session.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def get_data(self) -> dict[str, str]:
    """Get session data from `iscsiadm -m session -r <sid> -S`.

    Returns a plain dict rather than a typed model on purpose: the set of
    keys varies with the negotiated iSCSI session/connection parameters
    (see `PARAM_MAP` in `.parameters`), so a fixed model would need every
    key declared optional — no safer than a dict, and it would make
    callers that iterate all keys (e.g. `get_parameters()`) more awkward.
    This mirrors the NVMe JSON methods, which keep the same raw-dict
    exception for the same reason.
    """
    result = IscsiAdm().session(**{'-r': self.session_id, '-S': None})
    if result.failed:
        return {}

    data: dict[str, str] = {}
    for line in result.stdout.splitlines():
        if line and not line.startswith('#'):
            key_val = line.split(' = ', 1)
            if len(key_val) == 2:
                data[key_val[0]] = key_val[1]
    return data

get_data_p2()

Get session data with print level 2.

Source code in sts_libs/src/sts/iscsi/session.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
def get_data_p2(self) -> dict[str, str]:
    """Get session data with print level 2."""
    result = IscsiAdm().session(**{'-r': self.session_id, '-S': None, '-P': '2'})
    if result.failed:
        return {}

    data: dict[str, str] = {}
    for line in result.stdout.splitlines():
        if line and ': ' in line:
            key_val = line.replace('\t', '').split(': ', 1)
            if len(key_val) == 2:
                data[key_val[0]] = key_val[1]
    return data

get_disks()

Get list of disks attached to session.

Source code in sts_libs/src/sts/iscsi/session.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def get_disks(self) -> list[SessionDisk]:
    """Get list of disks attached to session."""
    result = IscsiAdm().session(**{'-r': self.session_id, '-P': '3'})
    if result.failed or 'Attached scsi disk' not in result.stdout:
        return []

    disks: list[SessionDisk] = []
    scsi_pattern = r'scsi(\d+)\s+Channel\s+(\d+)\s+Id\s+(\d+)\s+Lun:\s+(\d+)'
    disk_pattern = r'Attached\s+scsi\s+disk\s+(\w+)\s+State:\s+(\w+)'

    lines = result.stdout.splitlines()
    for i, line in enumerate(lines):
        scsi_match = re.search(scsi_pattern, line)
        if not scsi_match or i + 1 >= len(lines):
            continue

        disk_match = re.search(disk_pattern, lines[i + 1])
        if not disk_match:
            continue

        disks.append(
            SessionDisk(
                name=disk_match.group(1),
                state=disk_match.group(2),
                scsi_n=scsi_match.group(1),
                channel=scsi_match.group(2),
                id=scsi_match.group(3),
                lun=scsi_match.group(4),
            ),
        )

    return disks

get_parameters()

Get negotiated parameters from session.

Source code in sts_libs/src/sts/iscsi/session.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def get_parameters(self) -> dict[str, str]:
    """Get negotiated parameters from session."""
    data = self.get_data_p2()
    if not data:
        logger.warning('Failed to get session data')
        return {}

    negotiated: dict[str, str] = {}
    for param_name in PARAM_MAP:
        if param_name not in data:
            logger.warning(f'Parameter {param_name} not found in session data')
            continue
        negotiated[param_name] = data[param_name]

    return negotiated

logout()

Log out from session.

Source code in sts_libs/src/sts/iscsi/session.py 
42
43
44
45
46
47
48
def logout(self) -> bool:
    """Log out from session."""
    result = IscsiAdm().session_logout(self.session_id)
    if result.failed:
        logger.error('Logout failed')
        return False
    return True

SessionDisk pydantic-model

Bases: ReportModel

SCSI disk exposed through an iSCSI session (parsed from -P 3 output).

Show JSON schema: 
{
  "description": "SCSI disk exposed through an iSCSI session (parsed from -P 3 output).",
  "properties": {
    "name": {
      "title": "Name",
      "type": "string"
    },
    "state": {
      "title": "State",
      "type": "string"
    },
    "scsi_n": {
      "title": "Scsi N",
      "type": "string"
    },
    "channel": {
      "title": "Channel",
      "type": "string"
    },
    "id": {
      "title": "Id",
      "type": "string"
    },
    "lun": {
      "title": "Lun",
      "type": "string"
    }
  },
  "required": [
    "name",
    "state",
    "scsi_n",
    "channel",
    "id",
    "lun"
  ],
  "title": "SessionDisk",
  "type": "object"
}

Fields:

  • name (str)
  • state (str)
  • scsi_n (str)
  • channel (str)
  • id (str)
  • lun (str)
Source code in sts_libs/src/sts/iscsi/session.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
class SessionDisk(ReportModel):
    """SCSI disk exposed through an iSCSI session (parsed from -P 3 output)."""

    name: str
    state: str
    scsi_n: str
    channel: str
    id: str
    lun: str

    def is_running(self) -> bool:
        """Check if disk state is 'running'."""
        return self.state == 'running'

is_running()

Check if disk state is 'running'.

Source code in sts_libs/src/sts/iscsi/session.py 
30
31
32
def is_running(self) -> bool:
    """Check if disk state is 'running'."""
    return self.state == 'running'

Configuration

sts.iscsi.config

iSCSI configuration models and initiator name management.

IscsiConfig pydantic-model

Bases: StsBaseModel

iSCSI configuration.

Show JSON schema: 
{
  "$defs": {
    "IscsiInterface": {
      "additionalProperties": false,
      "description": "iSCSI interface configuration.",
      "properties": {
        "iscsi_ifacename": {
          "minLength": 1,
          "title": "Iscsi Ifacename",
          "type": "string"
        },
        "ipaddress": {
          "title": "Ipaddress",
          "type": "string"
        },
        "hwaddress": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Hwaddress"
        }
      },
      "required": [
        "iscsi_ifacename",
        "ipaddress"
      ],
      "title": "IscsiInterface",
      "type": "object"
    },
    "IscsiNode": {
      "additionalProperties": false,
      "description": "An open-iscsi node record \u2014 a portal on a target.\n\nPer the iscsiadm manpage, open-iscsi uses \"node\" to refer to\na portal on a target \u2014 identified by targetname + portal + interface.",
      "properties": {
        "target_iqn": {
          "title": "Target Iqn",
          "type": "string"
        },
        "portal": {
          "title": "Portal",
          "type": "string"
        },
        "iface": {
          "default": "default",
          "title": "Iface",
          "type": "string"
        }
      },
      "required": [
        "target_iqn",
        "portal"
      ],
      "title": "IscsiNode",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "iSCSI configuration.",
  "properties": {
    "initiatorname": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Initiatorname"
    },
    "ifaces": {
      "items": {
        "$ref": "#/$defs/IscsiInterface"
      },
      "title": "Ifaces",
      "type": "array"
    },
    "targets": {
      "anyOf": [
        {
          "items": {
            "$ref": "#/$defs/IscsiNode"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Targets"
    },
    "driver": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Driver"
    }
  },
  "title": "IscsiConfig",
  "type": "object"
}

Fields:

Source code in sts_libs/src/sts/iscsi/config.py 
71
72
73
74
75
76
77
class IscsiConfig(StsBaseModel):
    """iSCSI configuration."""

    initiatorname: str | None = None
    ifaces: list[IscsiInterface] = Field(default_factory=list)
    targets: list[IscsiNode] | None = None
    driver: str | None = None

IscsiInterface pydantic-model

Bases: StsBaseModel

iSCSI interface configuration.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "iSCSI interface configuration.",
  "properties": {
    "iscsi_ifacename": {
      "minLength": 1,
      "title": "Iscsi Ifacename",
      "type": "string"
    },
    "ipaddress": {
      "title": "Ipaddress",
      "type": "string"
    },
    "hwaddress": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Hwaddress"
    }
  },
  "required": [
    "iscsi_ifacename",
    "ipaddress"
  ],
  "title": "IscsiInterface",
  "type": "object"
}

Fields:

  • iscsi_ifacename (str)
  • ipaddress (str)
  • hwaddress (str | None)
Source code in sts_libs/src/sts/iscsi/config.py 
63
64
65
66
67
68
class IscsiInterface(StsBaseModel):
    """iSCSI interface configuration."""

    iscsi_ifacename: str = Field(min_length=1)
    ipaddress: str
    hwaddress: str | None = None

IscsidConfig

Bases: Config

Manages /etc/iscsi/iscsid.conf settings.

Source code in sts_libs/src/sts/iscsi/config.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class IscsidConfig(Config):
    """Manages /etc/iscsi/iscsid.conf settings."""

    CONFIG_PATH = Path('/etc/iscsi/iscsid.conf')

    def __init__(self) -> None:
        super().__init__(self.CONFIG_PATH)

    def set_chap(
        self,
        username: str,
        password: str,
        mutual_username: str | None = None,
        mutual_password: str | None = None,
    ) -> bool:
        """Configure CHAP authentication in iscsid.conf and restart iscsid."""
        if not username or not password:
            logger.error('Username and password required')
            return False

        params = {
            'node.session.auth.authmethod': 'CHAP',
            'node.session.auth.username': username,
            'node.session.auth.password': password,
            'discovery.sendtargets.auth.authmethod': 'CHAP',
            'discovery.sendtargets.auth.username': username,
            'discovery.sendtargets.auth.password': password,
        }
        if mutual_username and mutual_password:
            params.update(
                {
                    'node.session.auth.username_in': mutual_username,
                    'node.session.auth.password_in': mutual_password,
                    'discovery.sendtargets.auth.username_in': mutual_username,
                    'discovery.sendtargets.auth.password_in': mutual_password,
                }
            )

        self.set_parameters(params)
        if not self.save():
            return False
        if not SystemManager().service_restart(ISCSID_SERVICE_NAME):
            return False
        wait_for_iscsid()
        return True

    def disable_chap(self) -> bool:
        """Remove all CHAP settings from iscsid.conf and restart iscsid."""
        chap_keys = {
            'node.session.auth.authmethod',
            'node.session.auth.username',
            'node.session.auth.password',
            'node.session.auth.username_in',
            'node.session.auth.password_in',
            'discovery.sendtargets.auth.authmethod',
            'discovery.sendtargets.auth.username',
            'discovery.sendtargets.auth.password',
            'discovery.sendtargets.auth.username_in',
            'discovery.sendtargets.auth.password_in',
        }
        self.remove_parameters(chap_keys)
        if not self.save():
            return False
        return self._restart_iscsid()

    @staticmethod
    def _restart_iscsid() -> bool:
        """Restart iscsid and confirm it becomes responsive."""
        if not SystemManager().service_restart(ISCSID_SERVICE_NAME):
            return False
        return wait_for_iscsid()

disable_chap()

Remove all CHAP settings from iscsid.conf and restart iscsid.

Source code in sts_libs/src/sts/iscsi/config.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def disable_chap(self) -> bool:
    """Remove all CHAP settings from iscsid.conf and restart iscsid."""
    chap_keys = {
        'node.session.auth.authmethod',
        'node.session.auth.username',
        'node.session.auth.password',
        'node.session.auth.username_in',
        'node.session.auth.password_in',
        'discovery.sendtargets.auth.authmethod',
        'discovery.sendtargets.auth.username',
        'discovery.sendtargets.auth.password',
        'discovery.sendtargets.auth.username_in',
        'discovery.sendtargets.auth.password_in',
    }
    self.remove_parameters(chap_keys)
    if not self.save():
        return False
    return self._restart_iscsid()

set_chap(username, password, mutual_username=None, mutual_password=None)

Configure CHAP authentication in iscsid.conf and restart iscsid.

Source code in sts_libs/src/sts/iscsi/config.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def set_chap(
    self,
    username: str,
    password: str,
    mutual_username: str | None = None,
    mutual_password: str | None = None,
) -> bool:
    """Configure CHAP authentication in iscsid.conf and restart iscsid."""
    if not username or not password:
        logger.error('Username and password required')
        return False

    params = {
        'node.session.auth.authmethod': 'CHAP',
        'node.session.auth.username': username,
        'node.session.auth.password': password,
        'discovery.sendtargets.auth.authmethod': 'CHAP',
        'discovery.sendtargets.auth.username': username,
        'discovery.sendtargets.auth.password': password,
    }
    if mutual_username and mutual_password:
        params.update(
            {
                'node.session.auth.username_in': mutual_username,
                'node.session.auth.password_in': mutual_password,
                'discovery.sendtargets.auth.username_in': mutual_username,
                'discovery.sendtargets.auth.password_in': mutual_password,
            }
        )

    self.set_parameters(params)
    if not self.save():
        return False
    if not SystemManager().service_restart(ISCSID_SERVICE_NAME):
        return False
    wait_for_iscsid()
    return True

rand_iscsi_string(length)

Generate random string using iSCSI-allowed characters (RFC 7143 Section 6.1).

Source code in sts_libs/src/sts/iscsi/config.py 
29
30
31
def rand_iscsi_string(length: int) -> str | None:
    """Generate random string using iSCSI-allowed characters (RFC 7143 Section 6.1)."""
    return rand_string(length, chars=ISCSI_ALLOWED_CHARS)

set_initiatorname(name)

Write initiator name to /etc/iscsi/initiatorname.iscsi and restart iscsid.

Source code in sts_libs/src/sts/iscsi/config.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
def set_initiatorname(name: str) -> bool:
    """Write initiator name to /etc/iscsi/initiatorname.iscsi and restart iscsid."""
    system = SystemManager()
    try:
        Path('/etc/iscsi/initiatorname.iscsi').write_text(f'InitiatorName={name}\n')
        if not system.service_restart(ISCSID_SERVICE_NAME):
            logger.error('Failed to restart iscsid')
            return False
    except OSError:
        logger.exception('Failed to set initiator name')
        return False

    return wait_for_iscsid()

wait_for_iscsid()

Verify iscsid is responsive after a restart.

iscsid is a Type=notify systemd service, so systemctl restart already blocks until READY=1. This check confirms iscsiadm can talk to the daemon — a lightweight sanity gate rather than a polling loop.

Source code in sts_libs/src/sts/iscsi/config.py 
34
35
36
37
38
39
40
41
42
43
44
45
def wait_for_iscsid() -> bool:
    """Verify iscsid is responsive after a restart.

    iscsid is a Type=notify systemd service, so systemctl restart already
    blocks until READY=1.  This check confirms iscsiadm can talk to the
    daemon — a lightweight sanity gate rather than a polling loop.
    """
    result = IscsiAdm().session()
    if result.succeeded or 'No active sessions' in (result.stdout + result.stderr):
        return True
    logger.warning('iscsid not responding after restart')
    return False

iSCSI Administration

sts.iscsi.iscsiadm

Wrapper for the iscsiadm command line tool.

IscsiAdm pydantic-model

Bases: StsBaseModel

Wrapper for iscsiadm command line tool.

Attributes:

Name Type Description
debug_level Literal[0, 1, 2, 3, 4, 5, 6, 7, 8]

Debug verbosity level passed to iscsiadm via --debug.
Show JSON schema: 
{
  "additionalProperties": false,
  "description": "Wrapper for iscsiadm command line tool.\n\nAttributes:\n    debug_level: Debug verbosity level passed to iscsiadm via ``--debug``.",
  "properties": {
    "debug_level": {
      "default": 0,
      "enum": [
        0,
        1,
        2,
        3,
        4,
        5,
        6,
        7,
        8
      ],
      "title": "Debug Level",
      "type": "integer"
    }
  },
  "title": "IscsiAdm",
  "type": "object"
}

Fields:

  • debug_level (Literal[0, 1, 2, 3, 4, 5, 6, 7, 8])
Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class IscsiAdm(StsBaseModel):
    """Wrapper for iscsiadm command line tool.

    Attributes:
        debug_level: Debug verbosity level passed to iscsiadm via ``--debug``.
    """

    CLI_NAME: ClassVar[str] = 'iscsiadm'

    debug_level: Literal[0, 1, 2, 3, 4, 5, 6, 7, 8] = 0

    def _run(self, mode: str = '', arguments: dict[str, str | None] | None = None) -> CommandResult:
        """Build and execute an iscsiadm command."""
        command_list: list[str] = [self.CLI_NAME, '--mode', mode]
        if arguments is not None:
            command_list.extend(
                f'{k} {v}' if v is not None else k for k, v in arguments.items() if v is not None or k.startswith('-')
            )
        if self.debug_level:
            command_list += ['--debug', str(self.debug_level)]
        return run(' '.join(command_list))

    def iface(self, op: str, iface: str, name: str | None = None, value: str | None = None) -> CommandResult:
        """Run iscsiadm iface command."""
        arguments: dict[str, str | None] = {'-o': op, '-I': iface}
        if name is not None:
            arguments['-n'] = name
        if value is not None:
            arguments['-v'] = value
        return self._run(mode='iface', arguments=arguments)

    def iface_update(self, iface: str, name: str, value: str) -> CommandResult:
        """Update iSCSI interface parameter."""
        return self.iface(op='update', iface=iface, name=f'iface.{name}', value=value)

    def iface_update_iqn(self, iface: str, iqn: str) -> CommandResult:
        """Update iSCSI interface initiator name."""
        return self.iface_update(iface=iface, name='initiatorname', value=iqn)

    def iface_update_ip(self, iface: str, ip: str) -> CommandResult:
        """Update iSCSI interface IP address."""
        return self.iface_update(iface=iface, name='ipaddress', value=ip)

    def iface_exists(self, iface: str) -> bool:
        """Check if iSCSI interface exists."""
        return self.iface(op='show', iface=iface).succeeded

    def discovery(
        self,
        portal: str = '127.0.0.1',
        type: str = 'st',  # noqa: A002
        interface: str | None = None,
        **kwargs: str,
    ) -> CommandResult:
        """Run SendTargets discovery."""
        arguments: dict[str, str | None] = {'-t': type, '-p': portal}
        if kwargs:
            arguments.update(kwargs)
        if interface:
            arguments['-I'] = interface
        return self._run(mode='discovery', arguments=arguments)

    def node(self, **kwargs: str | None) -> CommandResult:
        """Run iscsiadm node command."""
        return self._run(mode='node', arguments=kwargs)

    def node_login(self, **kwargs: str) -> CommandResult:
        """Log in to a node."""
        arguments: dict[str, str | None] = {'--login': None}
        arguments.update(kwargs)
        return self.node(**arguments)

    def node_logout(self, **kwargs: str) -> CommandResult:
        """Log out from a node."""
        arguments: dict[str, str | None] = {'--logout': None}
        arguments.update(kwargs)
        return self.node(**arguments)

    def node_logoutall(self, how: Literal['all', 'manual', 'automatic', 'onboot'] = 'all') -> CommandResult:
        """Log out from all nodes."""
        return self.node(**{'--logoutall': how})

    def session(self, **kwargs: str | None) -> CommandResult:
        """Run iscsiadm session command."""
        return self._run(mode='session', arguments=kwargs)

    def session_logout(self, sid: str) -> CommandResult:
        """Log out a session by its ID."""
        return self.session(**{'-r': sid, '--logout': None})

discovery(portal='127.0.0.1', type='st', interface=None, **kwargs)

Run SendTargets discovery.

Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def discovery(
    self,
    portal: str = '127.0.0.1',
    type: str = 'st',  # noqa: A002
    interface: str | None = None,
    **kwargs: str,
) -> CommandResult:
    """Run SendTargets discovery."""
    arguments: dict[str, str | None] = {'-t': type, '-p': portal}
    if kwargs:
        arguments.update(kwargs)
    if interface:
        arguments['-I'] = interface
    return self._run(mode='discovery', arguments=arguments)

iface(op, iface, name=None, value=None)

Run iscsiadm iface command.

Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
39
40
41
42
43
44
45
46
def iface(self, op: str, iface: str, name: str | None = None, value: str | None = None) -> CommandResult:
    """Run iscsiadm iface command."""
    arguments: dict[str, str | None] = {'-o': op, '-I': iface}
    if name is not None:
        arguments['-n'] = name
    if value is not None:
        arguments['-v'] = value
    return self._run(mode='iface', arguments=arguments)

iface_exists(iface)

Check if iSCSI interface exists.

Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
60
61
62
def iface_exists(self, iface: str) -> bool:
    """Check if iSCSI interface exists."""
    return self.iface(op='show', iface=iface).succeeded

iface_update(iface, name, value)

Update iSCSI interface parameter.

Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
48
49
50
def iface_update(self, iface: str, name: str, value: str) -> CommandResult:
    """Update iSCSI interface parameter."""
    return self.iface(op='update', iface=iface, name=f'iface.{name}', value=value)

iface_update_ip(iface, ip)

Update iSCSI interface IP address.

Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
56
57
58
def iface_update_ip(self, iface: str, ip: str) -> CommandResult:
    """Update iSCSI interface IP address."""
    return self.iface_update(iface=iface, name='ipaddress', value=ip)

iface_update_iqn(iface, iqn)

Update iSCSI interface initiator name.

Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
52
53
54
def iface_update_iqn(self, iface: str, iqn: str) -> CommandResult:
    """Update iSCSI interface initiator name."""
    return self.iface_update(iface=iface, name='initiatorname', value=iqn)

node(**kwargs)

Run iscsiadm node command.

Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
79
80
81
def node(self, **kwargs: str | None) -> CommandResult:
    """Run iscsiadm node command."""
    return self._run(mode='node', arguments=kwargs)

node_login(**kwargs)

Log in to a node.

Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
83
84
85
86
87
def node_login(self, **kwargs: str) -> CommandResult:
    """Log in to a node."""
    arguments: dict[str, str | None] = {'--login': None}
    arguments.update(kwargs)
    return self.node(**arguments)

node_logout(**kwargs)

Log out from a node.

Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
89
90
91
92
93
def node_logout(self, **kwargs: str) -> CommandResult:
    """Log out from a node."""
    arguments: dict[str, str | None] = {'--logout': None}
    arguments.update(kwargs)
    return self.node(**arguments)

node_logoutall(how='all')

Log out from all nodes.

Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
95
96
97
def node_logoutall(self, how: Literal['all', 'manual', 'automatic', 'onboot'] = 'all') -> CommandResult:
    """Log out from all nodes."""
    return self.node(**{'--logoutall': how})

session(**kwargs)

Run iscsiadm session command.

Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
 99
100
101
def session(self, **kwargs: str | None) -> CommandResult:
    """Run iscsiadm session command."""
    return self._run(mode='session', arguments=kwargs)

session_logout(sid)

Log out a session by its ID.

Source code in sts_libs/src/sts/iscsi/iscsiadm.py 
103
104
105
def session_logout(self, sid: str) -> CommandResult:
    """Log out a session by its ID."""
    return self.session(**{'-r': sid, '--logout': None})

sts.iscsi.node

iSCSI node record (target + portal + iface).

IscsiNode pydantic-model

Bases: StsBaseModel

An open-iscsi node record — a portal on a target.

Per the iscsiadm manpage, open-iscsi uses "node" to refer to a portal on a target — identified by targetname + portal + interface.

Show JSON schema: 
{
  "additionalProperties": false,
  "description": "An open-iscsi node record \u2014 a portal on a target.\n\nPer the iscsiadm manpage, open-iscsi uses \"node\" to refer to\na portal on a target \u2014 identified by targetname + portal + interface.",
  "properties": {
    "target_iqn": {
      "title": "Target Iqn",
      "type": "string"
    },
    "portal": {
      "title": "Portal",
      "type": "string"
    },
    "iface": {
      "default": "default",
      "title": "Iface",
      "type": "string"
    }
  },
  "required": [
    "target_iqn",
    "portal"
  ],
  "title": "IscsiNode",
  "type": "object"
}

Fields:

  • target_iqn (str)
  • portal (str)
  • iface (str)
Source code in sts_libs/src/sts/iscsi/node.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class IscsiNode(StsBaseModel):
    """An open-iscsi node record — a portal on a target.

    Per the iscsiadm manpage, open-iscsi uses "node" to refer to
    a portal on a target — identified by targetname + portal + interface.
    """

    target_iqn: str
    portal: str
    iface: str = 'default'

    def login(self) -> CommandResult:
        """Log in to target and wait for SCSI devices to settle."""
        iscsiadm = IscsiAdm()
        result = iscsiadm.discovery(portal=self.portal)
        result.assert_ok('Discovery failed')

        result = iscsiadm.node_login(**{'-p': self.portal, '-T': self.target_iqn})
        result.assert_ok('Login failed')

        udevadm_settle()
        return result

    def logout(self) -> CommandResult:
        """Log out from target."""
        return IscsiAdm().node_logout(**{'-p': self.portal, '-T': self.target_iqn})

login()

Log in to target and wait for SCSI devices to settle.

Source code in sts_libs/src/sts/iscsi/node.py 
29
30
31
32
33
34
35
36
37
38
39
def login(self) -> CommandResult:
    """Log in to target and wait for SCSI devices to settle."""
    iscsiadm = IscsiAdm()
    result = iscsiadm.discovery(portal=self.portal)
    result.assert_ok('Discovery failed')

    result = iscsiadm.node_login(**{'-p': self.portal, '-T': self.target_iqn})
    result.assert_ok('Login failed')

    udevadm_settle()
    return result

logout()

Log out from target.

Source code in sts_libs/src/sts/iscsi/node.py 
41
42
43
def logout(self) -> CommandResult:
    """Log out from target."""
    return IscsiAdm().node_logout(**{'-p': self.portal, '-T': self.target_iqn})

sts.iscsi.parameters

iSCSI parameter negotiation verification per RFC 7143.

verify_parameter(param, target_value, initiator_value, negotiated_value, max_burst_length=None)

Verify negotiated parameter value against RFC 7143 rules.

Parameters:

Name Type Description Default
param str

Parameter name

 required
target_value str

Target's offered value

 required
initiator_value str

Initiator's offered value

 required
negotiated_value str

Actually negotiated value

 required
max_burst_length tuple[str, str] | None

(target, initiator) MaxBurstLength values for FirstBurstLength validation

 None
Source code in sts_libs/src/sts/iscsi/parameters.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def verify_parameter(
    param: str,
    target_value: str,
    initiator_value: str,
    negotiated_value: str,
    max_burst_length: tuple[str, str] | None = None,
) -> bool:
    """Verify negotiated parameter value against RFC 7143 rules.

    Args:
        param: Parameter name
        target_value: Target's offered value
        initiator_value: Initiator's offered value
        negotiated_value: Actually negotiated value
        max_burst_length: (target, initiator) MaxBurstLength values for FirstBurstLength validation
    """
    try:
        if param == 'HeaderDigest':
            expected = _verify_header_digest(target_value, initiator_value)

        elif param == 'MaxRecvDataSegmentLength':
            # Receiver controls its buffer size
            expected = initiator_value

        elif param in {'MaxXmitDataSegmentLength', 'MaxBurstLength'}:
            # Use minimum of offered values for data sizes
            expected = target_value if int(target_value) < int(initiator_value) else initiator_value

        elif param == 'ImmediateData':
            # Both must support immediate data for it to be enabled
            expected = 'Yes' if target_value == 'Yes' == initiator_value else 'No'

        elif param == 'InitialR2T':
            # Both must support disabled R2T for it to be disabled
            expected = 'No' if target_value == 'No' == initiator_value else 'Yes'

        elif param == 'FirstBurstLength':
            expected = _verify_first_burst_length(target_value, initiator_value, max_burst_length)

        else:
            logger.warning(f'Unknown parameter: {param}')
            return False

    except (ValueError, TypeError):
        logger.exception(f'Parameter validation error for {param}')
        return False

    # Log validation failures with details
    if expected != negotiated_value:
        logger.warning(
            f"""
            Parameter {param} validation failed:
            target={target_value}
            initiator={initiator_value}
            negotiated={negotiated_value} expected={expected}
            """
        )
        return False

    return True