Skip to content

ark_sm_service

ArkSMService

Bases: ArkService

Source code in ark_sdk_python/services/sm/ark_sm_service.py 
 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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
class ArkSMService(ArkService):
    def __init__(self, isp_auth: ArkISPAuth) -> None:
        super().__init__(isp_auth)
        self.__isp_auth = isp_auth
        self.__client: ArkISPServiceClient = ArkISPServiceClient.from_isp_auth(self.__isp_auth, 'sessionmonitoring')

    @validate_arguments
    def __search_params_from_filter(self, sessions_filter: ArkSMSessionsFilter):
        return {'search': sessions_filter.search}

    @validate_arguments
    def __call_sessions_api(self, params: Optional[dict] = None) -> ArkSMSessions:
        params_dict = {}
        if params:
            params_dict['params'] = params
        resp = self.__client.get(SESSIONS_API_URL, **params_dict)
        if resp.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to list sessions [{resp.text}] {params=}')
        return ArkSMSessions.parse_obj(resp.json())

    @validate_arguments
    def __call_activities_api(self, session_id: str, params: Optional[dict] = None) -> ArkSMSessionActivities:
        endpoint = SESSION_ACTIVITIES_API_URL.format(session_id=session_id)
        resp = self.__client.get(endpoint, params=params)
        if resp.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to list activities [{resp.text}]')
        return ArkSMSessionActivities.parse_obj(resp.json())

    @validate_arguments
    def __list_sessions(self, params: Optional[Dict] = None) -> Iterator[ArkSMPage]:
        params = params or {}
        sessions: ArkSMSessions = self.__call_sessions_api(params)
        offset = 0
        while sessions.returned_count > 0:
            yield ArkSMPage(items=sessions.sessions)
            offset += sessions.returned_count
            params['offset'] = offset
            sessions = self.__call_sessions_api(params)

    @validate_arguments
    def __list_activities(self, session_id: str, params: Optional[Dict] = None) -> Iterator[ArkSMActivitiesPage]:
        params = params or {}
        activities: ArkSMSessionActivities = self.__call_activities_api(session_id=session_id, params=params)
        offset = 0
        while activities.returned_count > 0:
            yield ArkSMActivitiesPage(items=activities.activities)
            offset += activities.returned_count
            params['offset'] = offset
            activities = self.__call_activities_api(session_id=session_id, params=params)

    def list_sessions(self) -> Iterator[ArkSMPage]:
        """
        Lists all sessions done on the last 24 hours

        Raises:
            ArkServiceException: _description_

        Yields:
            Iterator[ArkSMPage]: _description_
        """
        self._logger.info('Listing all session')
        yield from self.__list_sessions()

    def count_sessions(self) -> int:
        """
        Counts all sessions done on the last 24 hours

        Returns:
            int: _description_
        """
        return self.__call_sessions_api().filtered_count

    def list_sessions_by(self, sessions_filter: ArkSMSessionsFilter) -> Iterator[ArkSMPage]:
        """
        Lists all sessions with given filter

        Args:
            sessions_filter (ArkSMSessionsFilter): _description_
        Examples:
            ArkSMSessionsFilter(search='startTime GE 2023-12-03T08:55:29Z AND sessionDuration GE 00:00:01')
            ArkSMSessionsFilter(search='sessionStatus IN Failed,Ended AND endReason STARTSWITH Err008')
            ArkSMSessionsFilter(search='command STARTSWITH ls')
            ArkSMSessionsFilter(search='protocol IN SSH,RDP,Database')

        Raises:
            ArkServiceException: _description_

        Yields:
            Iterator[ArkSMPage]: _description_
        """
        self._logger.info('Listing sessions by filter', search=sessions_filter.search)
        yield from self.__list_sessions(self.__search_params_from_filter(sessions_filter))

    def count_sessions_by(self, sessions_filter: ArkSMSessionsFilter) -> int:
        """
        Counts all sessions with given filter

        Args:
            sessions_filter (ArkSMSessionsFilter): _description_
        Examples:
            ArkSMSessionsFilter(search='startTime GE 2023-12-03T08:55:29Z AND sessionDuration GE 00:00:01')
            ArkSMSessionsFilter(search='sessionStatus IN Failed,Ended AND endReason STARTSWITH Err008')
            ArkSMSessionsFilter(search='command STARTSWITH ls')
            ArkSMSessionsFilter(search='protocol IN SSH,RDP,Database')

        Returns:
            int: _description_
        """
        return self.__call_sessions_api(self.__search_params_from_filter(sessions_filter)).filtered_count

    def session(self, get_session: ArkSMGetSession) -> ArkSMSession:
        """
        Retrieves a session by id

        Args:
            get_session (ArkSMGetSession): _description_

        Raises:
            ArkServiceException: _description_
            ArkServiceException: _description_

        Returns:
            ArkSMSession: _description_
        """
        self._logger.info(f'Retrieving session by id [{get_session.session_id}]')
        resp = self.__client.get(SESSION_API_URL.format(session_id=get_session.session_id))
        if resp.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to list sessions [{resp.text}]')
        session = resp.json()
        if len(session) == 0:
            raise ArkServiceException(f'No session found for requested session id [{get_session.session_id}]')
        return ArkSMSession.parse_obj(session)

    def list_session_activities(self, get_session_activities: ArkSMGetSessionActivities) -> Iterator[ArkSMActivitiesPage]:
        """
        Lists all session activities by session id

        Args:
            get_session_activities (ArkSMGetSessionActivities): _description_

        Yields:
            Iterator[ArkSMActivitiesPage]: _description_
        """
        self._logger.info(f'Retrieving session activities by id [{get_session_activities.session_id}]')
        yield from self.__list_activities(session_id=get_session_activities.session_id)

    def count_session_activities(self, get_session_activities: ArkSMGetSessionActivities) -> int:
        """
        Count all session activities by session id

        Args:
            get_session_activities (ArkSMGetSessionActivities): _description_

        Returns:
            int: _description_
        """
        self._logger.info(f'Counting session activities by id [{get_session_activities.session_id}]')
        return self.__call_activities_api(session_id=get_session_activities.session_id).filtered_count

    def list_session_activities_by(self, session_activities_filter: ArkSMSessionActivitiesFilter) -> Iterator[ArkSMActivitiesPage]:
        """
        Lists all session activities for session id by filter

        Args:
            session_activities_filter (ArkSMSessionActivitiesFilter): _description_

        Yields:
            Iterator[ArkSMActivitiesPage]: _description_
        """
        self._logger.info(f'Retrieving session activities by id [{session_activities_filter.session_id}]')
        for page in self.__list_activities(session_id=session_activities_filter.session_id):
            yield ArkSMActivitiesPage(
                items=[activity for activity in page.items if session_activities_filter.command_contains in activity.command]
            )

    def count_session_activities_by(self, session_activities_filter: ArkSMSessionActivitiesFilter) -> int:
        """
        Count all session activities for session id by filter

        Args:
            session_activities_filter (ArkSMSessionActivitiesFilter): _description_

        Returns:
            int: _description_
        """
        count = 0
        self._logger.info(f'Counting session activities by id [{session_activities_filter.session_id}] and filter')
        for page in self.list_session_activities_by(session_activities_filter):
            count += len(page.items)
        return count

    def sessions_stats(self) -> ArkSMSessionsStats:
        """
        Returns statistics about the sessions in the last 30 days

        Returns:
            ArkSMSessionsStats: _description_
        """
        self._logger.info('Calculating sessions stats for the last 30 days')
        start_time_from = (datetime.now() - timedelta(days=30)).isoformat(timespec='seconds') + 'Z'
        sessions = list(
            itertools.chain.from_iterable(
                [p.items for p in self.list_sessions_by(ArkSMSessionsFilter(search=f'startTime ge {start_time_from}'))]
            )
        )
        sessions_stats = ArkSMSessionsStats.construct()
        sessions_stats.sessions_count = len(sessions)
        sessions_stats.sessions_failure_count = len([s for s in sessions if s.session_status == ArkSMSessionStatus.FAILED])

        # Get sessions per application code
        app_codes: Set[ArkApplicationCode] = {s.application_code for s in sessions}
        sessions_stats.sessions_count_per_application_code = {
            ac: len([s for s in sessions if s.application_code == ac]) for ac in app_codes
        }

        # Get sessions per platform
        platforms: Set[ArkWorkspaceType] = {s.platform for s in sessions}
        sessions_stats.sessions_count_per_platform = {p: len([s for s in sessions if s.platform == p]) for p in platforms}

        # Get sessions per protocol
        protocols: Set[ArkProtocolType] = {s.protocol for s in sessions}
        sessions_stats.sessions_count_per_protocol = {p: len([s for s in sessions if s.protocol == p]) for p in protocols}

        # Get sessions per status
        statuses: Set[ArkSMSessionStatus] = {s.session_status for s in sessions}
        sessions_stats.sessions_count_per_status = {st: len([s for s in sessions if s.session_status == st]) for st in statuses}

        return sessions_stats

    @staticmethod
    @overrides
    def service_config() -> ArkServiceConfig:
        return SERVICE_CONFIG

count_session_activities(get_session_activities)

Count all session activities by session id

Parameters:

Name Type Description Default
get_session_activities ArkSMGetSessionActivities

description

 required

Returns:

Name Type Description
int int

description
Source code in ark_sdk_python/services/sm/ark_sm_service.py 
189
190
191
192
193
194
195
196
197
198
199
200
def count_session_activities(self, get_session_activities: ArkSMGetSessionActivities) -> int:
    """
    Count all session activities by session id

    Args:
        get_session_activities (ArkSMGetSessionActivities): _description_

    Returns:
        int: _description_
    """
    self._logger.info(f'Counting session activities by id [{get_session_activities.session_id}]')
    return self.__call_activities_api(session_id=get_session_activities.session_id).filtered_count

count_session_activities_by(session_activities_filter)

Count all session activities for session id by filter

Parameters:

Name Type Description Default
session_activities_filter ArkSMSessionActivitiesFilter

description

 required

Returns:

Name Type Description
int int

description
Source code in ark_sdk_python/services/sm/ark_sm_service.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def count_session_activities_by(self, session_activities_filter: ArkSMSessionActivitiesFilter) -> int:
    """
    Count all session activities for session id by filter

    Args:
        session_activities_filter (ArkSMSessionActivitiesFilter): _description_

    Returns:
        int: _description_
    """
    count = 0
    self._logger.info(f'Counting session activities by id [{session_activities_filter.session_id}] and filter')
    for page in self.list_session_activities_by(session_activities_filter):
        count += len(page.items)
    return count

count_sessions()

Counts all sessions done on the last 24 hours

Returns:

Name Type Description
int int

description
Source code in ark_sdk_python/services/sm/ark_sm_service.py 
106
107
108
109
110
111
112
113
def count_sessions(self) -> int:
    """
    Counts all sessions done on the last 24 hours

    Returns:
        int: _description_
    """
    return self.__call_sessions_api().filtered_count

count_sessions_by(sessions_filter)

Counts all sessions with given filter

Parameters:

Name Type Description Default
sessions_filter ArkSMSessionsFilter

description

 required

Returns:

Name Type Description
int int

description
Source code in ark_sdk_python/services/sm/ark_sm_service.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def count_sessions_by(self, sessions_filter: ArkSMSessionsFilter) -> int:
    """
    Counts all sessions with given filter

    Args:
        sessions_filter (ArkSMSessionsFilter): _description_
    Examples:
        ArkSMSessionsFilter(search='startTime GE 2023-12-03T08:55:29Z AND sessionDuration GE 00:00:01')
        ArkSMSessionsFilter(search='sessionStatus IN Failed,Ended AND endReason STARTSWITH Err008')
        ArkSMSessionsFilter(search='command STARTSWITH ls')
        ArkSMSessionsFilter(search='protocol IN SSH,RDP,Database')

    Returns:
        int: _description_
    """
    return self.__call_sessions_api(self.__search_params_from_filter(sessions_filter)).filtered_count

list_session_activities(get_session_activities)

Lists all session activities by session id

Parameters:

Name Type Description Default
get_session_activities ArkSMGetSessionActivities

description

 required

Yields:

Type Description
ArkSMActivitiesPage

Iterator[ArkSMActivitiesPage]: description
Source code in ark_sdk_python/services/sm/ark_sm_service.py 
176
177
178
179
180
181
182
183
184
185
186
187
def list_session_activities(self, get_session_activities: ArkSMGetSessionActivities) -> Iterator[ArkSMActivitiesPage]:
    """
    Lists all session activities by session id

    Args:
        get_session_activities (ArkSMGetSessionActivities): _description_

    Yields:
        Iterator[ArkSMActivitiesPage]: _description_
    """
    self._logger.info(f'Retrieving session activities by id [{get_session_activities.session_id}]')
    yield from self.__list_activities(session_id=get_session_activities.session_id)

list_session_activities_by(session_activities_filter)

Lists all session activities for session id by filter

Parameters:

Name Type Description Default
session_activities_filter ArkSMSessionActivitiesFilter

description

 required

Yields:

Type Description
ArkSMActivitiesPage

Iterator[ArkSMActivitiesPage]: description
Source code in ark_sdk_python/services/sm/ark_sm_service.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def list_session_activities_by(self, session_activities_filter: ArkSMSessionActivitiesFilter) -> Iterator[ArkSMActivitiesPage]:
    """
    Lists all session activities for session id by filter

    Args:
        session_activities_filter (ArkSMSessionActivitiesFilter): _description_

    Yields:
        Iterator[ArkSMActivitiesPage]: _description_
    """
    self._logger.info(f'Retrieving session activities by id [{session_activities_filter.session_id}]')
    for page in self.__list_activities(session_id=session_activities_filter.session_id):
        yield ArkSMActivitiesPage(
            items=[activity for activity in page.items if session_activities_filter.command_contains in activity.command]
        )

list_sessions()

Lists all sessions done on the last 24 hours

Raises:

Type Description
ArkServiceException

description

Yields:

Type Description
ArkSMPage

Iterator[ArkSMPage]: description
Source code in ark_sdk_python/services/sm/ark_sm_service.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def list_sessions(self) -> Iterator[ArkSMPage]:
    """
    Lists all sessions done on the last 24 hours

    Raises:
        ArkServiceException: _description_

    Yields:
        Iterator[ArkSMPage]: _description_
    """
    self._logger.info('Listing all session')
    yield from self.__list_sessions()

list_sessions_by(sessions_filter)

Lists all sessions with given filter

Parameters:

Name Type Description Default
sessions_filter ArkSMSessionsFilter

description

 required

Raises:

Type Description
ArkServiceException

description

Yields:

Type Description
ArkSMPage

Iterator[ArkSMPage]: description
Source code in ark_sdk_python/services/sm/ark_sm_service.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def list_sessions_by(self, sessions_filter: ArkSMSessionsFilter) -> Iterator[ArkSMPage]:
    """
    Lists all sessions with given filter

    Args:
        sessions_filter (ArkSMSessionsFilter): _description_
    Examples:
        ArkSMSessionsFilter(search='startTime GE 2023-12-03T08:55:29Z AND sessionDuration GE 00:00:01')
        ArkSMSessionsFilter(search='sessionStatus IN Failed,Ended AND endReason STARTSWITH Err008')
        ArkSMSessionsFilter(search='command STARTSWITH ls')
        ArkSMSessionsFilter(search='protocol IN SSH,RDP,Database')

    Raises:
        ArkServiceException: _description_

    Yields:
        Iterator[ArkSMPage]: _description_
    """
    self._logger.info('Listing sessions by filter', search=sessions_filter.search)
    yield from self.__list_sessions(self.__search_params_from_filter(sessions_filter))

session(get_session)

Retrieves a session by id

Parameters:

Name Type Description Default
get_session ArkSMGetSession

description

 required

Raises:

Type Description
ArkServiceException

description
ArkServiceException

description

Returns:

Name Type Description
ArkSMSession ArkSMSession

description
Source code in ark_sdk_python/services/sm/ark_sm_service.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def session(self, get_session: ArkSMGetSession) -> ArkSMSession:
    """
    Retrieves a session by id

    Args:
        get_session (ArkSMGetSession): _description_

    Raises:
        ArkServiceException: _description_
        ArkServiceException: _description_

    Returns:
        ArkSMSession: _description_
    """
    self._logger.info(f'Retrieving session by id [{get_session.session_id}]')
    resp = self.__client.get(SESSION_API_URL.format(session_id=get_session.session_id))
    if resp.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to list sessions [{resp.text}]')
    session = resp.json()
    if len(session) == 0:
        raise ArkServiceException(f'No session found for requested session id [{get_session.session_id}]')
    return ArkSMSession.parse_obj(session)

sessions_stats()

Returns statistics about the sessions in the last 30 days

Returns:

Name Type Description
ArkSMSessionsStats ArkSMSessionsStats

description
Source code in ark_sdk_python/services/sm/ark_sm_service.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def sessions_stats(self) -> ArkSMSessionsStats:
    """
    Returns statistics about the sessions in the last 30 days

    Returns:
        ArkSMSessionsStats: _description_
    """
    self._logger.info('Calculating sessions stats for the last 30 days')
    start_time_from = (datetime.now() - timedelta(days=30)).isoformat(timespec='seconds') + 'Z'
    sessions = list(
        itertools.chain.from_iterable(
            [p.items for p in self.list_sessions_by(ArkSMSessionsFilter(search=f'startTime ge {start_time_from}'))]
        )
    )
    sessions_stats = ArkSMSessionsStats.construct()
    sessions_stats.sessions_count = len(sessions)
    sessions_stats.sessions_failure_count = len([s for s in sessions if s.session_status == ArkSMSessionStatus.FAILED])

    # Get sessions per application code
    app_codes: Set[ArkApplicationCode] = {s.application_code for s in sessions}
    sessions_stats.sessions_count_per_application_code = {
        ac: len([s for s in sessions if s.application_code == ac]) for ac in app_codes
    }

    # Get sessions per platform
    platforms: Set[ArkWorkspaceType] = {s.platform for s in sessions}
    sessions_stats.sessions_count_per_platform = {p: len([s for s in sessions if s.platform == p]) for p in platforms}

    # Get sessions per protocol
    protocols: Set[ArkProtocolType] = {s.protocol for s in sessions}
    sessions_stats.sessions_count_per_protocol = {p: len([s for s in sessions if s.protocol == p]) for p in protocols}

    # Get sessions per status
    statuses: Set[ArkSMSessionStatus] = {s.session_status for s in sessions}
    sessions_stats.sessions_count_per_status = {st: len([s for s in sessions if s.session_status == st]) for st in statuses}

    return sessions_stats