iSCSI

This section documents the iSCSI (Internet Small Computer Systems Interface) functionality.

Device Management

sts.iscsi.device

iSCSI device management.

This module provides functionality for managing iSCSI devices: - Device discovery - Device information - Device operations

iSCSI (Internet SCSI) enables: - Block storage over IP networks - SAN functionality without FC hardware - Storage consolidation and sharing - Remote boot capabilities

Key concepts: - Initiator: Client that accesses storage - Target: Storage server that provides devices - IQN: iSCSI Qualified Name (unique identifier) - Portal: IP:Port where target listens - Session: Connection between initiator and target

IscsiDevice dataclass

iSCSI device representation.

An iSCSI device represents a remote block device that: - Appears as a local SCSI device - Requires network connectivity - Can use authentication (CHAP) - Supports multipath for redundancy

Parameters:

Name	Type	Description	Default
name	str	Device name (e.g. 'sda')	required
path	Path | str	Device path (e.g. '/dev/sda')	required
size	int	Device size in bytes	required
ip	str	Target IP address	required
port	int	Target port (default: 3260)	3260
target_iqn	str	Target IQN (e.g. 'iqn.2003-01.org.linux-iscsi.target')	required
initiator_iqn	str | None	Initiator IQN (optional, from /etc/iscsi/initiatorname.iscsi)	None

Example

device = IscsiDevice(
    name='sda',
    path='/dev/sda',
    size=1000000000000,
    ip='192.168.1.100',
    target_iqn='iqn.2003-01.org.linux-iscsi.target',
)

Source code in sts_libs/src/sts/iscsi/device.py

@dataclass
class IscsiDevice:
    """iSCSI device representation.

    An iSCSI device represents a remote block device that:
    - Appears as a local SCSI device
    - Requires network connectivity
    - Can use authentication (CHAP)
    - Supports multipath for redundancy

    Args:
        name: Device name (e.g. 'sda')
        path: Device path (e.g. '/dev/sda')
        size: Device size in bytes
        ip: Target IP address
        port: Target port (default: 3260)
        target_iqn: Target IQN (e.g. 'iqn.2003-01.org.linux-iscsi.target')
        initiator_iqn: Initiator IQN (optional, from /etc/iscsi/initiatorname.iscsi)

    Example:
        ```python
        device = IscsiDevice(
            name='sda',
            path='/dev/sda',
            size=1000000000000,
            ip='192.168.1.100',
            target_iqn='iqn.2003-01.org.linux-iscsi.target',
        )
        ```
    """

    name: str
    path: Path | str
    size: int
    ip: str
    target_iqn: str
    port: int = 3260
    initiator_iqn: str | None = None

    # Internal fields
    _session_id: str | None = field(init=False, default=None)
    _iscsiadm: IscsiAdm = field(init=False, default_factory=IscsiAdm)

    # Important system paths
    ISCSI_PATH: ClassVar[Path] = Path('/sys/class/iscsi_session')  # Session info
    DATABASE_PATH: ClassVar[Path] = Path('/var/lib/iscsi')  # iSCSI database
    CONFIG_PATH: ClassVar[Path] = Path('/etc/iscsi')  # Configuration

    def __post_init__(self) -> None:
        """Initialize iSCSI device.

        Converts path string to Path object if needed.
        """
        # Convert path to Path if needed
        if isinstance(self.path, str):
            self.path = Path(self.path)

    @property
    def session_id(self) -> str | None:
        """Get iSCSI session ID.

        The session ID uniquely identifies an active connection:
        - Format is typically a small integer (e.g. '1')
        - None if device is not logged in
        - Cached to avoid repeated lookups

        Returns:
            Session ID if device is logged in, None otherwise

        Example:
            ```python
            device.session_id
            '1'
            ```
        """
        if self._session_id:
            return self._session_id

        # Try to find session ID by matching target and portal
        for session in IscsiSession.get_all():
            if session.target_iqn == self.target_iqn and session.portal.startswith(self.ip):
                self._session_id = session.session_id
                return self._session_id

        return None

    @session_id.setter
    def session_id(self, value: str | None) -> None:
        """Set iSCSI session ID.

        Args:
            value: Session ID
        """
        self._session_id = value

    @session_id.deleter
    def session_id(self) -> None:
        """Delete iSCSI session ID.

        Clears cached session ID when session ends.
        """
        self._session_id = None

    def login(self) -> bool:
        """Log in to target.

        Login process:
        1. Discover target (SendTargets)
        2. Create session
        3. Authenticate if needed
        4. Create device nodes

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            device.login()
            True
            ```
        """
        if self.session_id:
            logging.info('Already logged in')
            return True

        # First discover the target using SendTargets
        result = self._iscsiadm.discovery(portal=f'{self.ip}:{self.port}')
        if result.failed:
            logging.error('Discovery failed')
            return False

        # Then login to create session
        result = self._iscsiadm.node_login(**{'-p': f'{self.ip}:{self.port}', '-T': self.target_iqn})
        if result.failed:
            logging.error('Login failed')
            return False

        return True

    def logout(self) -> bool:
        """Log out from target.

        Logout process:
        1. Close session
        2. Remove device nodes
        3. Clear session ID

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            device.logout()
            True
            ```
        """
        if not self.session_id:
            logging.info('Not logged in')
            return True

        result = self._iscsiadm.node_logout(**{'-p': f'{self.ip}:{self.port}', '-T': self.target_iqn})
        if result.failed:
            logging.error('Logout failed')
            return False

        self._session_id = None
        return True

    def _update_config_file(self, parameters: dict[str, str]) -> bool:
        """Update iscsid.conf with parameters.

        Updates configuration while preserving:
        - Comments and formatting
        - Existing settings
        - File permissions

        Args:
            parameters: Parameters to update

        Returns:
            True if successful, False otherwise
        """
        config_path = self.CONFIG_PATH / 'iscsid.conf'
        try:
            # Read existing config
            with config_path.open('r') as f:
                lines = f.readlines()

            # Update or add parameters
            for key, value in parameters.items():
                found = False
                for i, line in enumerate(lines):
                    if line.strip().startswith('#'):
                        continue
                    if '=' in line:
                        file_key = line.split('=', 1)[0].strip()
                        if file_key == key:
                            lines[i] = f'{key} = {value}\n'
                            found = True
                            break
                if not found:
                    lines.append(f'{key} = {value}\n')

            # Write updated config
            with config_path.open('w') as f:
                f.writelines(lines)
        except OSError:
            logging.exception('Failed to update config file')
            return False

        return True

    def _restart_service(self) -> bool:
        """Restart iscsid service.

        Required after configuration changes:
        - Reloads configuration
        - Maintains existing sessions
        - Updates authentication

        Returns:
            True if successful, False otherwise
        """
        return SystemManager().service_restart(ISCSID_SERVICE_NAME)

    def set_chap(
        self,
        username: str,
        password: str,
        mutual_username: str | None = None,
        mutual_password: str | None = None,
    ) -> bool:
        """Set CHAP authentication.

        CHAP (Challenge Handshake Authentication Protocol):
        - One-way: Target authenticates initiator
        - Mutual: Both sides authenticate each other
        - Requires matching config on target

        Args:
            username: CHAP username
            password: CHAP password
            mutual_username: Mutual CHAP username (optional)
            mutual_password: Mutual CHAP password (optional)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            device.set_chap('user', 'pass')
            True
            ```
        """
        if not username or not password:
            logging.error('Username and password required')
            return False

        # Build parameters for both node and discovery
        parameters = {
            '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,
        }

        # Add mutual CHAP if provided
        if mutual_username and mutual_password:
            parameters.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,
                }
            )

        # Update config and restart service
        return self._update_config_file(parameters) and self._restart_service()

    def disable_chap(self) -> bool:
        """Disable CHAP authentication.

        Removes all CHAP settings:
        - Node authentication
        - Discovery authentication
        - Mutual CHAP settings

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            device.disable_chap()
            True
            ```
        """
        config_path = self.CONFIG_PATH / 'iscsid.conf'
        try:
            # Read existing config
            with config_path.open('r') as f:
                lines = f.readlines()

            # Keep only non-CHAP lines
            new_lines = [
                line
                for line in lines
                if not any(
                    p in line
                    for p in (
                        'node.session.auth.',
                        'discovery.sendtargets.auth.',
                    )
                )
                or line.strip().startswith('#')
            ]

            # Write updated config
            with config_path.open('w') as f:
                f.writelines(new_lines)
        except OSError:
            logging.exception('Failed to update config file')
            return False

        return self._restart_service()

    @classmethod
    def _parse_device_info(cls, info: dict[str, str]) -> IscsiDevice | None:
        """Parse device info into IscsiDevice instance.

        Creates device object from dictionary:
        - Validates required fields
        - Converts types as needed
        - Handles missing fields

        Args:
            info: Device info dictionary

        Returns:
            IscsiDevice instance or None if parsing failed
        """
        try:
            return cls(
                name=info['name'],
                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):
            logging.warning('Failed to create device')
            return None

    @classmethod
    def get_all(cls) -> list[IscsiDevice]:
        """Get list of all iSCSI devices.

        Discovery process:
        1. Get all active sessions
        2. Get portal info for each session
        3. Get disk info for each session
        4. Create device objects

        Returns:
            List of IscsiDevice instances

        Example:
            ```python
            IscsiDevice.get_all()
            [IscsiDevice(name='sda', ...), IscsiDevice(name='sdb', ...)]
            ```
        """
        devices = []
        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=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 targets.

        Uses SendTargets discovery:
        - Queries portal for targets
        - Returns list of IQNs
        - No authentication needed
        - Fast operation

        Args:
            ip: Target IP address
            port: Target port (default: 3260)

        Returns:
            List of discovered target IQNs

        Example:
            ```python
            IscsiDevice.discover('192.168.1.100')
            ['iqn.2003-01.org.linux-iscsi.target1', 'iqn.2003-01.org.linux-iscsi.target2']
            ```
        """
        iscsiadm = IscsiAdm()
        result = iscsiadm.discovery(portal=f'{ip}:{port}')
        if result.failed:
            logging.warning('No targets found')
            return []

        targets = []
        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

session_id: str | None deletable property writable

Get iSCSI session ID.

The session ID uniquely identifies an active connection: - Format is typically a small integer (e.g. '1') - None if device is not logged in - Cached to avoid repeated lookups

Returns:

Type	Description
str | None	Session ID if device is logged in, None otherwise

Example

device.session_id
'1'

__post_init__()

Initialize iSCSI device.

Converts path string to Path object if needed.

Source code in sts_libs/src/sts/iscsi/device.py

def __post_init__(self) -> None:
    """Initialize iSCSI device.

    Converts path string to Path object if needed.
    """
    # Convert path to Path if needed
    if isinstance(self.path, str):
        self.path = Path(self.path)

disable_chap()

Disable CHAP authentication.

Removes all CHAP settings: - Node authentication - Discovery authentication - Mutual CHAP settings

Returns:

Type	Description
bool	True if successful, False otherwise

Example

device.disable_chap()
True

Source code in sts_libs/src/sts/iscsi/device.py

def disable_chap(self) -> bool:
    """Disable CHAP authentication.

    Removes all CHAP settings:
    - Node authentication
    - Discovery authentication
    - Mutual CHAP settings

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        device.disable_chap()
        True
        ```
    """
    config_path = self.CONFIG_PATH / 'iscsid.conf'
    try:
        # Read existing config
        with config_path.open('r') as f:
            lines = f.readlines()

        # Keep only non-CHAP lines
        new_lines = [
            line
            for line in lines
            if not any(
                p in line
                for p in (
                    'node.session.auth.',
                    'discovery.sendtargets.auth.',
                )
            )
            or line.strip().startswith('#')
        ]

        # Write updated config
        with config_path.open('w') as f:
            f.writelines(new_lines)
    except OSError:
        logging.exception('Failed to update config file')
        return False

    return self._restart_service()

discover(ip, port=3260) classmethod

Discover available targets.

Uses SendTargets discovery: - Queries portal for targets - Returns list of IQNs - No authentication needed - Fast operation

Parameters:

Name	Type	Description	Default
ip	str	Target IP address	required
port	int	Target port (default: 3260)	3260

Returns:

Type	Description
list[str]	List of discovered target IQNs

Example

IscsiDevice.discover('192.168.1.100')
['iqn.2003-01.org.linux-iscsi.target1', 'iqn.2003-01.org.linux-iscsi.target2']

Source code in sts_libs/src/sts/iscsi/device.py

@classmethod
def discover(cls, ip: str, port: int = 3260) -> list[str]:
    """Discover available targets.

    Uses SendTargets discovery:
    - Queries portal for targets
    - Returns list of IQNs
    - No authentication needed
    - Fast operation

    Args:
        ip: Target IP address
        port: Target port (default: 3260)

    Returns:
        List of discovered target IQNs

    Example:
        ```python
        IscsiDevice.discover('192.168.1.100')
        ['iqn.2003-01.org.linux-iscsi.target1', 'iqn.2003-01.org.linux-iscsi.target2']
        ```
    """
    iscsiadm = IscsiAdm()
    result = iscsiadm.discovery(portal=f'{ip}:{port}')
    if result.failed:
        logging.warning('No targets found')
        return []

    targets = []
    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 list of all iSCSI devices.

Discovery process: 1. Get all active sessions 2. Get portal info for each session 3. Get disk info for each session 4. Create device objects

Returns:

Type	Description
list[IscsiDevice]	List of IscsiDevice instances

Example

IscsiDevice.get_all()
[IscsiDevice(name='sda', ...), IscsiDevice(name='sdb', ...)]

Source code in sts_libs/src/sts/iscsi/device.py

@classmethod
def get_all(cls) -> list[IscsiDevice]:
    """Get list of all iSCSI devices.

    Discovery process:
    1. Get all active sessions
    2. Get portal info for each session
    3. Get disk info for each session
    4. Create device objects

    Returns:
        List of IscsiDevice instances

    Example:
        ```python
        IscsiDevice.get_all()
        [IscsiDevice(name='sda', ...), IscsiDevice(name='sdb', ...)]
        ```
    """
    devices = []
    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=f'/dev/{disk.name}',
                size=size,
                ip=ip,
                port=int(port),
                target_iqn=session.target_iqn,
            )
            devices.append(device)

    return devices

login()

Log in to target.

Login process: 1. Discover target (SendTargets) 2. Create session 3. Authenticate if needed 4. Create device nodes

Returns:

Type	Description
bool	True if successful, False otherwise

Example

device.login()
True

Source code in sts_libs/src/sts/iscsi/device.py

def login(self) -> bool:
    """Log in to target.

    Login process:
    1. Discover target (SendTargets)
    2. Create session
    3. Authenticate if needed
    4. Create device nodes

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        device.login()
        True
        ```
    """
    if self.session_id:
        logging.info('Already logged in')
        return True

    # First discover the target using SendTargets
    result = self._iscsiadm.discovery(portal=f'{self.ip}:{self.port}')
    if result.failed:
        logging.error('Discovery failed')
        return False

    # Then login to create session
    result = self._iscsiadm.node_login(**{'-p': f'{self.ip}:{self.port}', '-T': self.target_iqn})
    if result.failed:
        logging.error('Login failed')
        return False

    return True

logout()

Log out from target.

Logout process: 1. Close session 2. Remove device nodes 3. Clear session ID

Returns:

Type	Description
bool	True if successful, False otherwise

Example

device.logout()
True

Source code in sts_libs/src/sts/iscsi/device.py

def logout(self) -> bool:
    """Log out from target.

    Logout process:
    1. Close session
    2. Remove device nodes
    3. Clear session ID

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        device.logout()
        True
        ```
    """
    if not self.session_id:
        logging.info('Not logged in')
        return True

    result = self._iscsiadm.node_logout(**{'-p': f'{self.ip}:{self.port}', '-T': self.target_iqn})
    if result.failed:
        logging.error('Logout failed')
        return False

    self._session_id = None
    return True

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

Set CHAP authentication.

CHAP (Challenge Handshake Authentication Protocol): - One-way: Target authenticates initiator - Mutual: Both sides authenticate each other - Requires matching config on target

Parameters:

Name	Type	Description	Default
username	str	CHAP username	required
password	str	CHAP password	required
mutual_username	str | None	Mutual CHAP username (optional)	None
mutual_password	str | None	Mutual CHAP password (optional)	None

Returns:

Type	Description
bool	True if successful, False otherwise

Example

device.set_chap('user', 'pass')
True

Source code in sts_libs/src/sts/iscsi/device.py

def set_chap(
    self,
    username: str,
    password: str,
    mutual_username: str | None = None,
    mutual_password: str | None = None,
) -> bool:
    """Set CHAP authentication.

    CHAP (Challenge Handshake Authentication Protocol):
    - One-way: Target authenticates initiator
    - Mutual: Both sides authenticate each other
    - Requires matching config on target

    Args:
        username: CHAP username
        password: CHAP password
        mutual_username: Mutual CHAP username (optional)
        mutual_password: Mutual CHAP password (optional)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        device.set_chap('user', 'pass')
        True
        ```
    """
    if not username or not password:
        logging.error('Username and password required')
        return False

    # Build parameters for both node and discovery
    parameters = {
        '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,
    }

    # Add mutual CHAP if provided
    if mutual_username and mutual_password:
        parameters.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,
            }
        )

    # Update config and restart service
    return self._update_config_file(parameters) and self._restart_service()

Session Management

sts.iscsi.session

iSCSI session management.

This module provides functionality for managing iSCSI sessions: - Session discovery - Session information - Session operations

An iSCSI session represents: - Connection between initiator and target - Authentication and parameters - Associated SCSI devices - Network transport details

IscsiSession dataclass

iSCSI session representation.

A session maintains the connection state between: - Initiator (local system) - Target (remote storage) - Multiple connections possible - Negotiated parameters

Parameters:

Name	Type	Description	Default
session_id	str	Session ID (unique identifier)	required
target_iqn	str	Target IQN (iSCSI Qualified Name)	required
portal	str	Portal address (IP:Port)	required

Source code in sts_libs/src/sts/iscsi/session.py

@dataclass
class IscsiSession:
    """iSCSI session representation.

    A session maintains the connection state between:
    - Initiator (local system)
    - Target (remote storage)
    - Multiple connections possible
    - Negotiated parameters

    Args:
        session_id: Session ID (unique identifier)
        target_iqn: Target IQN (iSCSI Qualified Name)
        portal: Portal address (IP:Port)
    """

    session_id: str
    target_iqn: str
    portal: str
    _iscsiadm: IscsiAdm = field(init=False, default_factory=IscsiAdm)

    def logout(self) -> bool:
        """Log out from session.

        Logout process:
        - Closes all connections
        - Removes session
        - Cleans up resources

        Returns:
            True if successful, False otherwise
        """
        result = self._iscsiadm.node_logout(
            **{'-p': self.portal, '-T': self.target_iqn},
        )
        if result.failed:
            logging.error('Logout failed')
            return False
        return True

    def get_data(self) -> dict[str, str]:
        """Get session data.

        Retrieves basic session information:
        - Authentication method
        - CHAP credentials
        - Connection parameters
        - Session options

        Returns:
            Dictionary of session data from 'iscsiadm -m session -r <sid> -S'

        Example:
            ```python
            session.get_data()
            {
                'node.session.auth.authmethod': 'None',
                'node.session.auth.username': '',
                'node.session.auth.password': '',
                ...
            }
            ```
        """
        result = self._iscsiadm.session(**{'-r': self.session_id, '-S': None})
        if result.failed:
            return {}

        # Parse key=value pairs, skipping comments
        data = {}
        lines = [line for line in result.stdout.splitlines() if line and not line.startswith('#')]
        for line in lines:
            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.

        Retrieves detailed session information:
        - Target details
        - Portal information
        - Connection state
        - SCSI information

        Returns:
            Dictionary of session data from 'iscsiadm -m session -r <sid> -S -P 2'

        Example:
            ```python
            session.get_data_p2()
            {
                'Target': 'iqn.2003-01.org.linux-iscsi.target',
                'Current Portal': '192.168.1.100:3260,1',
                'Persistent Portal': '192.168.1.100:3260',
                ...
            }
            ```
        """
        result = self._iscsiadm.session(**{'-r': self.session_id, '-S': None, '-P': '2'})
        if result.failed:
            return {}

        # Parse key: value pairs, removing tabs
        data = {}
        lines = [line.replace('\t', '') for line in result.stdout.splitlines() if line and ': ' in line]
        for line in lines:
            key_val = line.split(': ', 1)
            if len(key_val) == 2:
                data[key_val[0]] = key_val[1]
        return data

    @dataclass
    class SessionDisk:
        """iSCSI session disk representation.

        Represents a SCSI disk exposed through the session:
        - Maps to local block device
        - Has SCSI addressing (H:C:T:L)
        - Tracks operational state

        Attributes:
            name: Disk name (e.g. 'sda')
            state: Disk state (e.g. 'running')
            scsi_n: SCSI host number
            channel: SCSI channel number
            id: SCSI target ID
            lun: SCSI Logical Unit Number
        """

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

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

            'running' state indicates:
            - Device is accessible
            - I/O is possible
            - No error conditions

            Returns:
                True if disk state is 'running'
            """
            return self.state == 'running'

    def get_disks(self) -> list[SessionDisk]:
        """Get list of disks attached to session.

        Discovers SCSI disks by:
        - Parsing session information
        - Matching SCSI addresses
        - Checking device states

        Returns:
            List of SessionDisk instances

        Example:
            ```python
            session.get_disks()
            [SessionDisk(name='sda', state='running', scsi_n='2', channel='00', id='0', lun='0')]
            ```
        """
        result = self._iscsiadm.session(**{'-r': self.session_id, '-P': '3'})
        if result.failed or 'Attached scsi disk' not in result.stdout:
            return []

        disks = []
        # Match SCSI info lines like: "scsi2 Channel 00 Id 0 Lun: 0"
        scsi_pattern = r'scsi(\d+)\s+Channel\s+(\d+)\s+Id\s+(\d+)\s+Lun:\s+(\d+)'
        # Match disk info lines like: "Attached scsi disk sda State: running"
        disk_pattern = r'Attached\s+scsi\s+disk\s+(\w+)\s+State:\s+(\w+)'

        lines = result.stdout.splitlines()
        for i, line in enumerate(lines):
            # First find SCSI address
            scsi_match = re.search(scsi_pattern, line)
            if not scsi_match:
                continue

            # Then look for disk info in next line
            if i + 1 >= len(lines):
                continue

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

            # Create disk object with all information
            disks.append(
                self.SessionDisk(
                    name=disk_match.group(1),  # Device name (e.g. sda)
                    state=disk_match.group(2),  # Device state
                    scsi_n=scsi_match.group(1),  # Host number
                    channel=scsi_match.group(2),  # Channel number
                    id=scsi_match.group(3),  # Target ID
                    lun=scsi_match.group(4),  # LUN
                ),
            )

        return disks

    @classmethod
    def get_all(cls) -> list[IscsiSession]:
        """Get list of all iSCSI sessions.

        Lists active sessions by:
        - Querying iscsiadm
        - Parsing session information
        - Creating session objects

        Returns:
            List of IscsiSession instances
        """
        iscsiadm = IscsiAdm()
        result = iscsiadm.session()
        if result.failed:
            return []

        sessions = []
        for line in result.stdout.splitlines():
            # Parse line like:
            # tcp: [1] 192.168.1.100:3260,1 iqn.2003-01.target
            if not line:
                continue
            parts = line.split()
            if len(parts) < 4:  # Need at least tcp: [id] portal target
                continue
            session_id = parts[1].strip('[]')
            portal = parts[2].split(',')[0]  # Remove portal group tag
            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 session by target IQN.

        Finds session matching target:
        - Searches all active sessions
        - Matches exact IQN
        - Returns all matches

        Args:
            target_iqn: Target IQN

        Returns:
            List of IscsiSession instances matching target IQN
        """
        return [session for session in cls.get_all() if session.target_iqn == target_iqn]

    @classmethod
    def get_by_portal(cls, portal: str) -> list[IscsiSession]:
        """Get session by portal address.

        Finds session matching portal:
        - Searches all active sessions
        - Matches exact portal address
        - Returns first match

        Args:
            portal: Portal address (e.g. '127.0.0.1:3260')

        Returns:
            List of IscsiSession instances matching portal
        """
        return [session for session in cls.get_all() if session.portal == portal]

    def get_parameters(self) -> dict[str, str]:
        """Get parameters from session.

        Retrieves negotiated parameters:
        - Connection settings
        - Authentication options
        - Performance tuning
        - Error recovery

        Returns:
            Dictionary of parameters
        """
        from .parameters import PARAM_MAP  # Parameter name mapping

        # Get parameters from session data
        data = self.get_data_p2()
        if not data:
            logging.warning('Failed to get session data')
            return {}

        # Extract negotiated parameters
        negotiated = {}
        for param_name in PARAM_MAP:
            if param_name not in data:
                logging.warning(f'Parameter {param_name} not found in session data')
                continue
            negotiated[param_name] = data[param_name]

        return negotiated

SessionDisk dataclass

iSCSI session disk representation.

Represents a SCSI disk exposed through the session: - Maps to local block device - Has SCSI addressing (H:C:T:L) - Tracks operational state

Attributes:

Name	Type	Description
name	str	Disk name (e.g. 'sda')
state	str	Disk state (e.g. 'running')
scsi_n	str	SCSI host number
channel	str	SCSI channel number
id	str	SCSI target ID
lun	str	SCSI Logical Unit Number

Source code in sts_libs/src/sts/iscsi/session.py

@dataclass
class SessionDisk:
    """iSCSI session disk representation.

    Represents a SCSI disk exposed through the session:
    - Maps to local block device
    - Has SCSI addressing (H:C:T:L)
    - Tracks operational state

    Attributes:
        name: Disk name (e.g. 'sda')
        state: Disk state (e.g. 'running')
        scsi_n: SCSI host number
        channel: SCSI channel number
        id: SCSI target ID
        lun: SCSI Logical Unit Number
    """

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

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

        'running' state indicates:
        - Device is accessible
        - I/O is possible
        - No error conditions

        Returns:
            True if disk state is 'running'
        """
        return self.state == 'running'

is_running()

Check if disk is running.

'running' state indicates: - Device is accessible - I/O is possible - No error conditions

Returns:

Type	Description
bool	True if disk state is 'running'

Source code in sts_libs/src/sts/iscsi/session.py

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

    'running' state indicates:
    - Device is accessible
    - I/O is possible
    - No error conditions

    Returns:
        True if disk state is 'running'
    """
    return self.state == 'running'

get_all() classmethod

Get list of all iSCSI sessions.

Lists active sessions by: - Querying iscsiadm - Parsing session information - Creating session objects

Returns:

Type	Description
list[IscsiSession]	List of IscsiSession instances

Source code in sts_libs/src/sts/iscsi/session.py

@classmethod
def get_all(cls) -> list[IscsiSession]:
    """Get list of all iSCSI sessions.

    Lists active sessions by:
    - Querying iscsiadm
    - Parsing session information
    - Creating session objects

    Returns:
        List of IscsiSession instances
    """
    iscsiadm = IscsiAdm()
    result = iscsiadm.session()
    if result.failed:
        return []

    sessions = []
    for line in result.stdout.splitlines():
        # Parse line like:
        # tcp: [1] 192.168.1.100:3260,1 iqn.2003-01.target
        if not line:
            continue
        parts = line.split()
        if len(parts) < 4:  # Need at least tcp: [id] portal target
            continue
        session_id = parts[1].strip('[]')
        portal = parts[2].split(',')[0]  # Remove portal group tag
        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 session by portal address.

Finds session matching portal: - Searches all active sessions - Matches exact portal address - Returns first match

Parameters:

Name	Type	Description	Default
portal	str	Portal address (e.g. '127.0.0.1:3260')	required

Returns:

Type	Description
list[IscsiSession]	List of IscsiSession instances matching portal

Source code in sts_libs/src/sts/iscsi/session.py

@classmethod
def get_by_portal(cls, portal: str) -> list[IscsiSession]:
    """Get session by portal address.

    Finds session matching portal:
    - Searches all active sessions
    - Matches exact portal address
    - Returns first match

    Args:
        portal: Portal address (e.g. '127.0.0.1:3260')

    Returns:
        List of IscsiSession instances matching portal
    """
    return [session for session in cls.get_all() if session.portal == portal]

get_by_target(target_iqn) classmethod

Get session by target IQN.

Finds session matching target: - Searches all active sessions - Matches exact IQN - Returns all matches

Parameters:

Name	Type	Description	Default
target_iqn	str	Target IQN	required

Returns:

Type	Description
list[IscsiSession]	List of IscsiSession instances matching target IQN

Source code in sts_libs/src/sts/iscsi/session.py

@classmethod
def get_by_target(cls, target_iqn: str) -> list[IscsiSession]:
    """Get session by target IQN.

    Finds session matching target:
    - Searches all active sessions
    - Matches exact IQN
    - Returns all matches

    Args:
        target_iqn: Target IQN

    Returns:
        List of IscsiSession instances matching target IQN
    """
    return [session for session in cls.get_all() if session.target_iqn == target_iqn]

get_data()

Get session data.

Retrieves basic session information: - Authentication method - CHAP credentials - Connection parameters - Session options

Returns:

Type	Description
dict[str, str]	Dictionary of session data from 'iscsiadm -m session -r -S'

Example

session.get_data()
{
    'node.session.auth.authmethod': 'None',
    'node.session.auth.username': '',
    'node.session.auth.password': '',
    ...
}

Source code in sts_libs/src/sts/iscsi/session.py

def get_data(self) -> dict[str, str]:
    """Get session data.

    Retrieves basic session information:
    - Authentication method
    - CHAP credentials
    - Connection parameters
    - Session options

    Returns:
        Dictionary of session data from 'iscsiadm -m session -r <sid> -S'

    Example:
        ```python
        session.get_data()
        {
            'node.session.auth.authmethod': 'None',
            'node.session.auth.username': '',
            'node.session.auth.password': '',
            ...
        }
        ```
    """
    result = self._iscsiadm.session(**{'-r': self.session_id, '-S': None})
    if result.failed:
        return {}

    # Parse key=value pairs, skipping comments
    data = {}
    lines = [line for line in result.stdout.splitlines() if line and not line.startswith('#')]
    for line in lines:
        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.

Retrieves detailed session information: - Target details - Portal information - Connection state - SCSI information

Returns:

Type	Description
dict[str, str]	Dictionary of session data from 'iscsiadm -m session -r -S -P 2'

Example

session.get_data_p2()
{
    'Target': 'iqn.2003-01.org.linux-iscsi.target',
    'Current Portal': '192.168.1.100:3260,1',
    'Persistent Portal': '192.168.1.100:3260',
    ...
}

Source code in sts_libs/src/sts/iscsi/session.py

def get_data_p2(self) -> dict[str, str]:
    """Get session data with print level 2.

    Retrieves detailed session information:
    - Target details
    - Portal information
    - Connection state
    - SCSI information

    Returns:
        Dictionary of session data from 'iscsiadm -m session -r <sid> -S -P 2'

    Example:
        ```python
        session.get_data_p2()
        {
            'Target': 'iqn.2003-01.org.linux-iscsi.target',
            'Current Portal': '192.168.1.100:3260,1',
            'Persistent Portal': '192.168.1.100:3260',
            ...
        }
        ```
    """
    result = self._iscsiadm.session(**{'-r': self.session_id, '-S': None, '-P': '2'})
    if result.failed:
        return {}

    # Parse key: value pairs, removing tabs
    data = {}
    lines = [line.replace('\t', '') for line in result.stdout.splitlines() if line and ': ' in line]
    for line in lines:
        key_val = line.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.

Discovers SCSI disks by: - Parsing session information - Matching SCSI addresses - Checking device states

Returns:

Type	Description
list[SessionDisk]	List of SessionDisk instances

Example

session.get_disks()
[SessionDisk(name='sda', state='running', scsi_n='2', channel='00', id='0', lun='0')]

Source code in sts_libs/src/sts/iscsi/session.py

def get_disks(self) -> list[SessionDisk]:
    """Get list of disks attached to session.

    Discovers SCSI disks by:
    - Parsing session information
    - Matching SCSI addresses
    - Checking device states

    Returns:
        List of SessionDisk instances

    Example:
        ```python
        session.get_disks()
        [SessionDisk(name='sda', state='running', scsi_n='2', channel='00', id='0', lun='0')]
        ```
    """
    result = self._iscsiadm.session(**{'-r': self.session_id, '-P': '3'})
    if result.failed or 'Attached scsi disk' not in result.stdout:
        return []

    disks = []
    # Match SCSI info lines like: "scsi2 Channel 00 Id 0 Lun: 0"
    scsi_pattern = r'scsi(\d+)\s+Channel\s+(\d+)\s+Id\s+(\d+)\s+Lun:\s+(\d+)'
    # Match disk info lines like: "Attached scsi disk sda State: running"
    disk_pattern = r'Attached\s+scsi\s+disk\s+(\w+)\s+State:\s+(\w+)'

    lines = result.stdout.splitlines()
    for i, line in enumerate(lines):
        # First find SCSI address
        scsi_match = re.search(scsi_pattern, line)
        if not scsi_match:
            continue

        # Then look for disk info in next line
        if i + 1 >= len(lines):
            continue

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

        # Create disk object with all information
        disks.append(
            self.SessionDisk(
                name=disk_match.group(1),  # Device name (e.g. sda)
                state=disk_match.group(2),  # Device state
                scsi_n=scsi_match.group(1),  # Host number
                channel=scsi_match.group(2),  # Channel number
                id=scsi_match.group(3),  # Target ID
                lun=scsi_match.group(4),  # LUN
            ),
        )

    return disks

get_parameters()

Get parameters from session.

Retrieves negotiated parameters: - Connection settings - Authentication options - Performance tuning - Error recovery

Returns:

Type	Description
dict[str, str]	Dictionary of parameters

Source code in sts_libs/src/sts/iscsi/session.py

def get_parameters(self) -> dict[str, str]:
    """Get parameters from session.

    Retrieves negotiated parameters:
    - Connection settings
    - Authentication options
    - Performance tuning
    - Error recovery

    Returns:
        Dictionary of parameters
    """
    from .parameters import PARAM_MAP  # Parameter name mapping

    # Get parameters from session data
    data = self.get_data_p2()
    if not data:
        logging.warning('Failed to get session data')
        return {}

    # Extract negotiated parameters
    negotiated = {}
    for param_name in PARAM_MAP:
        if param_name not in data:
            logging.warning(f'Parameter {param_name} not found in session data')
            continue
        negotiated[param_name] = data[param_name]

    return negotiated

logout()

Log out from session.

Logout process: - Closes all connections - Removes session - Cleans up resources

Returns:

Type	Description
bool	True if successful, False otherwise

Source code in sts_libs/src/sts/iscsi/session.py

def logout(self) -> bool:
    """Log out from session.

    Logout process:
    - Closes all connections
    - Removes session
    - Cleans up resources

    Returns:
        True if successful, False otherwise
    """
    result = self._iscsiadm.node_logout(
        **{'-p': self.portal, '-T': self.target_iqn},
    )
    if result.failed:
        logging.error('Logout failed')
        return False
    return True

Configuration

sts.iscsi.config

iSCSI configuration.

This module provides configuration classes for iSCSI: - Interface configuration - Node configuration - Overall configuration - Daemon configuration

Key components: - Initiator name (unique identifier) - Network interfaces - Target discovery - Authentication - Service management

ConfVars

Bases: TypedDict

iSCSI configuration variables.

Complete configuration includes: - Local initiator name - List of targets to connect to - Network interface configurations - Hardware offload driver (if used)

Source code in sts_libs/src/sts/iscsi/config.py

class ConfVars(TypedDict):
    """iSCSI configuration variables.

    Complete configuration includes:
    - Local initiator name
    - List of targets to connect to
    - Network interface configurations
    - Hardware offload driver (if used)
    """

    initiatorname: str  # iqn.1994-05.redhat:example
    targets: list[IscsiNode]  # Target configurations
    ifaces: list[IscsiInterface]  # Interface configurations
    driver: str | None  # Hardware offload driver (e.g. 'bnx2i', 'qedi', 'cxgb4i')

IscsiConfig dataclass

iSCSI configuration.

Complete iSCSI initiator configuration: - Local identification - Network setup - Target connections - Hardware offload

Attributes:

Name	Type	Description
initiatorname	str | None	Initiator IQN (local identifier)
ifaces	list[IscsiInterface]	List of interface configurations
targets	list[IscsiNode] | None	List of target configurations
driver	str | None	Hardware offload driver name (if used)

Source code in sts_libs/src/sts/iscsi/config.py

@dataclass
class IscsiConfig:
    """iSCSI configuration.

    Complete iSCSI initiator configuration:
    - Local identification
    - Network setup
    - Target connections
    - Hardware offload

    Attributes:
        initiatorname: Initiator IQN (local identifier)
        ifaces: List of interface configurations
        targets: List of target configurations
        driver: Hardware offload driver name (if used)
    """

    initiatorname: str | None
    ifaces: list[IscsiInterface]
    targets: list[IscsiNode] | None
    driver: str | None

IscsiInterface dataclass

iSCSI interface configuration.

An interface defines how iSCSI traffic is sent: - Which network interface to use - What IP address to bind to - Optional hardware address binding

Attributes:

Name	Type	Description
iscsi_ifacename	str	Interface name (e.g. 'iface0')
ipaddress	str	IP address to bind to
hwaddress	str | None	Hardware address (optional, for HBA binding)

Source code in sts_libs/src/sts/iscsi/config.py

@dataclass
class IscsiInterface:
    """iSCSI interface configuration.

    An interface defines how iSCSI traffic is sent:
    - Which network interface to use
    - What IP address to bind to
    - Optional hardware address binding

    Attributes:
        iscsi_ifacename: Interface name (e.g. 'iface0')
        ipaddress: IP address to bind to
        hwaddress: Hardware address (optional, for HBA binding)
    """

    iscsi_ifacename: str
    ipaddress: str
    hwaddress: str | None = None

IscsiNode dataclass

iSCSI node configuration.

A node represents a connection to a target: - Target identification - Network location - Interface binding - Session management

Attributes:

Name	Type	Description
target_iqn	str	Target IQN (unique identifier)
portal	str	Portal address (IP:Port)
interface	str	Interface name to use

Source code in sts_libs/src/sts/iscsi/config.py

@dataclass
class IscsiNode:
    """iSCSI node configuration.

    A node represents a connection to a target:
    - Target identification
    - Network location
    - Interface binding
    - Session management

    Attributes:
        target_iqn: Target IQN (unique identifier)
        portal: Portal address (IP:Port)
        interface: Interface name to use
    """

    target_iqn: str
    portal: str
    interface: str
    iscsiadm: IscsiAdm = field(init=False, default_factory=IscsiAdm)

    def login(self) -> bool:
        """Log in to target.

        Login process:
        1. Discover target if needed
        2. Create new session
        3. Authenticate if configured
        4. Create SCSI devices

        Returns:
            True if successful, False otherwise
        """
        # First discover the target
        result = self.iscsiadm.discovery(portal=self.portal)
        if result.failed:
            logging.error('Discovery failed')
            return False

        # Then login to create session
        result = self.iscsiadm.node_login(**{'-p': self.portal, '-T': self.target_iqn})
        if result.failed:
            logging.error('Login failed')
            return False
        return True

    def logout(self) -> bool:
        """Log out from target.

        Logout process:
        1. Close session
        2. Remove SCSI devices
        3. Clean up resources

        Returns:
            True if successful, False otherwise
        """
        result = self.iscsiadm.node_logout(**{'-p': self.portal, '-T': self.target_iqn})
        if result.failed:
            logging.error('Logout failed')
            return False
        return True

    @classmethod
    def setup_and_login(cls, portal: str, initiator_iqn: str, target_iqn: str | None = None) -> IscsiNode:
        """Set up node and log in.

        Complete setup process:
        1. Create node configuration
        2. Discover target if needed
        3. Log in to create session

        Args:
            portal: Portal address (IP:Port)
            initiator_iqn: Initiator IQN (local identifier)
            target_iqn: Target IQN (optional, will be discovered)

        Returns:
            IscsiNode instance
        """
        # Create node with known info
        node = cls(
            target_iqn=target_iqn or '',  # Will be discovered if not provided
            portal=portal,
            interface=initiator_iqn,
        )

        # Discover target if IQN not provided
        if not target_iqn:
            result = node.iscsiadm.discovery(portal=portal)
            if result.succeeded:
                for line in result.stdout.splitlines():
                    if line.strip():  # Skip empty lines
                        # Format: portal target_iqn
                        # Example: 192.168.1.100:3260,1 iqn.2003-01.org.linux-iscsi:target1
                        parts = line.split()
                        if len(parts) >= 2:
                            node.target_iqn = parts[1]
                            break

        # Log in if we have target IQN
        if node.target_iqn:
            node.login()

        return node

login()

Log in to target.

Login process: 1. Discover target if needed 2. Create new session 3. Authenticate if configured 4. Create SCSI devices

Returns:

Type	Description
bool	True if successful, False otherwise

Source code in sts_libs/src/sts/iscsi/config.py

def login(self) -> bool:
    """Log in to target.

    Login process:
    1. Discover target if needed
    2. Create new session
    3. Authenticate if configured
    4. Create SCSI devices

    Returns:
        True if successful, False otherwise
    """
    # First discover the target
    result = self.iscsiadm.discovery(portal=self.portal)
    if result.failed:
        logging.error('Discovery failed')
        return False

    # Then login to create session
    result = self.iscsiadm.node_login(**{'-p': self.portal, '-T': self.target_iqn})
    if result.failed:
        logging.error('Login failed')
        return False
    return True

logout()

Log out from target.

Logout process: 1. Close session 2. Remove SCSI devices 3. Clean up resources

Returns:

Type	Description
bool	True if successful, False otherwise

Source code in sts_libs/src/sts/iscsi/config.py

def logout(self) -> bool:
    """Log out from target.

    Logout process:
    1. Close session
    2. Remove SCSI devices
    3. Clean up resources

    Returns:
        True if successful, False otherwise
    """
    result = self.iscsiadm.node_logout(**{'-p': self.portal, '-T': self.target_iqn})
    if result.failed:
        logging.error('Logout failed')
        return False
    return True

setup_and_login(portal, initiator_iqn, target_iqn=None) classmethod

Set up node and log in.

Complete setup process: 1. Create node configuration 2. Discover target if needed 3. Log in to create session

Parameters:

Name	Type	Description	Default
portal	str	Portal address (IP:Port)	required
initiator_iqn	str	Initiator IQN (local identifier)	required
target_iqn	str | None	Target IQN (optional, will be discovered)	None

Returns:

Type	Description
IscsiNode	IscsiNode instance

Source code in sts_libs/src/sts/iscsi/config.py

@classmethod
def setup_and_login(cls, portal: str, initiator_iqn: str, target_iqn: str | None = None) -> IscsiNode:
    """Set up node and log in.

    Complete setup process:
    1. Create node configuration
    2. Discover target if needed
    3. Log in to create session

    Args:
        portal: Portal address (IP:Port)
        initiator_iqn: Initiator IQN (local identifier)
        target_iqn: Target IQN (optional, will be discovered)

    Returns:
        IscsiNode instance
    """
    # Create node with known info
    node = cls(
        target_iqn=target_iqn or '',  # Will be discovered if not provided
        portal=portal,
        interface=initiator_iqn,
    )

    # Discover target if IQN not provided
    if not target_iqn:
        result = node.iscsiadm.discovery(portal=portal)
        if result.succeeded:
            for line in result.stdout.splitlines():
                if line.strip():  # Skip empty lines
                    # Format: portal target_iqn
                    # Example: 192.168.1.100:3260,1 iqn.2003-01.org.linux-iscsi:target1
                    parts = line.split()
                    if len(parts) >= 2:
                        node.target_iqn = parts[1]
                        break

    # Log in if we have target IQN
    if node.target_iqn:
        node.login()

    return node

IscsidConfig

iSCSI daemon configuration.

Manages iscsid.conf settings: - Connection parameters - Authentication - Timeouts - Retry behavior

Source code in sts_libs/src/sts/iscsi/config.py

class IscsidConfig:
    """iSCSI daemon configuration.

    Manages iscsid.conf settings:
    - Connection parameters
    - Authentication
    - Timeouts
    - Retry behavior
    """

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

    def __init__(self) -> None:
        """Initialize configuration.

        Loads existing config if available.
        """
        self.parameters: dict[str, str] = {}
        if self.CONFIG_PATH.exists():
            self._load_config()

    def _load_config(self) -> None:
        """Load configuration from file.

        Parses iscsid.conf format:
        - Key = Value pairs
        - Comments start with #
        - Whitespace ignored
        """
        try:
            lines = self.CONFIG_PATH.read_text().splitlines()
            for line in lines:
                stripped_line = line.strip()
                # Skip empty lines and comments
                if not stripped_line or stripped_line.startswith('#'):
                    continue
                if '=' in stripped_line:
                    key, value = (s.strip() for s in stripped_line.split('=', 1))
                    self.parameters[key] = value
        except OSError:
            logging.exception('Failed to load config')

    def set_parameters(self, parameters: dict[str, str]) -> None:
        """Set configuration parameters.

        Updates multiple parameters at once:
        - Overwrites existing values
        - Adds new parameters
        - No validation performed

        Args:
            parameters: Dictionary of parameter names and values
        """
        self.parameters.update(parameters)

    def get_parameter(self, name: str) -> str | None:
        """Get parameter value.

        Args:
            name: Parameter name (dot-separated)

        Returns:
            Parameter value if found, None otherwise
        """
        return self.parameters.get(name)

    def save(self) -> bool:
        """Save configuration to file.

        Preserves file format:
        - Keeps comments
        - Maintains spacing
        - Updates values

        Returns:
            True if successful, False otherwise
        """
        try:
            # Read existing lines to preserve comments and formatting
            lines = []
            if self.CONFIG_PATH.exists():
                lines = self.CONFIG_PATH.read_text().splitlines()

            # Update or add parameters
            for key, value in self.parameters.items():
                found = False
                for i, line in enumerate(lines):
                    # Skip commented lines
                    if line.strip().startswith('#'):
                        continue
                    # Parse existing keys
                    if '=' in line:
                        file_key = line.split('=', 1)[0].strip()
                        if file_key == key:
                            lines[i] = f'{key} = {value}'
                            found = True
                            break
                # If key wasn't found, add it
                if not found:
                    lines.append(f'{key} = {value}')

            # Write back all lines
            self.CONFIG_PATH.write_text('\n'.join(lines) + '\n')
        except OSError:
            logging.exception('Failed to save config')
            return False
        return True

__init__()

Initialize configuration.

Loads existing config if available.

Source code in sts_libs/src/sts/iscsi/config.py

def __init__(self) -> None:
    """Initialize configuration.

    Loads existing config if available.
    """
    self.parameters: dict[str, str] = {}
    if self.CONFIG_PATH.exists():
        self._load_config()

get_parameter(name)

Get parameter value.

Parameters:

Name	Type	Description	Default
name	str	Parameter name (dot-separated)	required

Returns:

Type	Description
str | None	Parameter value if found, None otherwise

Source code in sts_libs/src/sts/iscsi/config.py

def get_parameter(self, name: str) -> str | None:
    """Get parameter value.

    Args:
        name: Parameter name (dot-separated)

    Returns:
        Parameter value if found, None otherwise
    """
    return self.parameters.get(name)

save()

Save configuration to file.

Preserves file format: - Keeps comments - Maintains spacing - Updates values

Returns:

Type	Description
bool	True if successful, False otherwise

Source code in sts_libs/src/sts/iscsi/config.py

def save(self) -> bool:
    """Save configuration to file.

    Preserves file format:
    - Keeps comments
    - Maintains spacing
    - Updates values

    Returns:
        True if successful, False otherwise
    """
    try:
        # Read existing lines to preserve comments and formatting
        lines = []
        if self.CONFIG_PATH.exists():
            lines = self.CONFIG_PATH.read_text().splitlines()

        # Update or add parameters
        for key, value in self.parameters.items():
            found = False
            for i, line in enumerate(lines):
                # Skip commented lines
                if line.strip().startswith('#'):
                    continue
                # Parse existing keys
                if '=' in line:
                    file_key = line.split('=', 1)[0].strip()
                    if file_key == key:
                        lines[i] = f'{key} = {value}'
                        found = True
                        break
            # If key wasn't found, add it
            if not found:
                lines.append(f'{key} = {value}')

        # Write back all lines
        self.CONFIG_PATH.write_text('\n'.join(lines) + '\n')
    except OSError:
        logging.exception('Failed to save config')
        return False
    return True

set_parameters(parameters)

Set configuration parameters.

Updates multiple parameters at once: - Overwrites existing values - Adds new parameters - No validation performed

Parameters:

Name	Type	Description	Default
parameters	dict[str, str]	Dictionary of parameter names and values	required

Source code in sts_libs/src/sts/iscsi/config.py

def set_parameters(self, parameters: dict[str, str]) -> None:
    """Set configuration parameters.

    Updates multiple parameters at once:
    - Overwrites existing values
    - Adds new parameters
    - No validation performed

    Args:
        parameters: Dictionary of parameter names and values
    """
    self.parameters.update(parameters)

create_iscsi_iface(iface_name)

Create iSCSI interface.

An iSCSI interface binds sessions to: - Network interface - IP address - Hardware address - Transport type

Parameters:

Name	Type	Description	Default
iface_name	str	Interface name	required

Returns:

Type	Description
bool	True if successful, False otherwise

Source code in sts_libs/src/sts/iscsi/config.py

def create_iscsi_iface(iface_name: str) -> bool:
    """Create iSCSI interface.

    An iSCSI interface binds sessions to:
    - Network interface
    - IP address
    - Hardware address
    - Transport type

    Args:
        iface_name: Interface name

    Returns:
        True if successful, False otherwise
    """
    iscsiadm = IscsiAdm()
    result = iscsiadm.iface(op='new', iface=iface_name)
    if result.failed:
        logging.error('Failed to create interface')
        return False
    return True

rand_iscsi_string(length)

Generate random string following iSCSI naming rules.

Text format is based on Internet Small Computer System Interface (iSCSI) Protocol (RFC 7143) Section 6.1: - ASCII letters and digits - Limited punctuation - No spaces or control chars

Parameters:

Name	Type	Description	Default
length	int	Length of the random string	required

Returns:

Type	Description
str | None	Random string or None if length is invalid

Example

rand_iscsi_string(8)
'Abc123_+'

Source code in sts_libs/src/sts/iscsi/config.py

def rand_iscsi_string(length: int) -> str | None:
    """Generate random string following iSCSI naming rules.

    Text format is based on Internet Small Computer System Interface (iSCSI) Protocol
    (RFC 7143) Section 6.1:
    - ASCII letters and digits
    - Limited punctuation
    - No spaces or control chars

    Args:
        length: Length of the random string

    Returns:
        Random string or None if length is invalid

    Example:
        ```python
        rand_iscsi_string(8)
        'Abc123_+'
        ```
    """
    return rand_string(length, chars=ISCSI_ALLOWED_CHARS)

set_initiatorname(name)

Set initiator name.

The initiator name uniquely identifies this system: - Must be valid IQN format - Must be unique on network - Stored in /etc/iscsi/initiatorname.iscsi

Parameters:

Name	Type	Description	Default
name	str	Initiator name (IQN format)	required

Returns:

Type	Description
bool	True if successful, False otherwise

Source code in sts_libs/src/sts/iscsi/config.py

def set_initiatorname(name: str) -> bool:
    """Set initiator name.

    The initiator name uniquely identifies this system:
    - Must be valid IQN format
    - Must be unique on network
    - Stored in /etc/iscsi/initiatorname.iscsi

    Args:
        name: Initiator name (IQN format)

    Returns:
        True if successful, False otherwise
    """
    system = SystemManager()
    try:
        Path('/etc/iscsi/initiatorname.iscsi').write_text(f'InitiatorName={name}\n')
        system.service_restart(ISCSID_SERVICE_NAME)
    except OSError:
        logging.exception('Failed to set initiator name')
        return False
    return True

setup(variables)

Configure iSCSI initiator based on env variables.

Complete setup process: 1. Set initiator name 2. Configure interfaces 3. Start required services 4. Discover targets 5. Enable services

Parameters:

Name	Type	Description	Default
variables	IscsiConfig	Configuration variables	required

Returns:

Type	Description
bool	True if successful, False otherwise

Source code in sts_libs/src/sts/iscsi/config.py

def setup(variables: IscsiConfig) -> bool:
    """Configure iSCSI initiator based on env variables.

    Complete setup process:
    1. Set initiator name
    2. Configure interfaces
    3. Start required services
    4. Discover targets
    5. Enable services

    Args:
        variables: Configuration variables

    Returns:
        True if successful, False otherwise
    """
    iscsiadm = IscsiAdm()
    system = SystemManager()

    # Set initiator name
    if variables.initiatorname and not set_initiatorname(variables.initiatorname):
        return False

    # Set up network interfaces
    if variables.ifaces:
        for iface in variables.ifaces:
            ifacename = iface.iscsi_ifacename
            # Enable iscsiuio for hardware offload
            if variables.driver in {'qedi', 'bnx2i'} and not system.is_service_running(ISCSIUIO_SERVICE_NAME):
                if not system.service_enable(ISCSIUIO_SERVICE_NAME):
                    return False
                if not system.service_start(ISCSIUIO_SERVICE_NAME):
                    return False

            # Create interface if needed
            if not iscsiadm.iface_exists(iface=ifacename) and not create_iscsi_iface(iface_name=ifacename):
                return False

            # Update interface parameters
            result = iscsiadm.iface_update(iface=ifacename, name='ipaddress', value=iface.ipaddress)
            if result.failed:
                logging.error(f'Failed to update interface: {result.stderr}')
                return False

            if iface.hwaddress:
                result = iscsiadm.iface_update(iface=ifacename, name='hwaddress', value=iface.hwaddress)
                if result.failed:
                    logging.error(f'Failed to update interface: {result.stderr}')
                    return False

    # Discover configured targets
    if variables.targets:
        for target in variables.targets:
            if not iscsiadm.discovery(
                portal=target.portal, interface=target.interface, target_iqn=target.target_iqn
            ).succeeded:
                return False

    # Enable iscsid service for automatic startup
    if not system.is_service_enabled(ISCSID_SERVICE_NAME):
        return system.service_enable(ISCSID_SERVICE_NAME)
    return True