Merge pull request #95 from RomanSemkin/fix-linter-complaints

Linter complaints were fixed

Author: rsemkin

examples/example-access-group-key.py

@@ -34,6 +34,7 @@
 # NOTE: fill in your credentials from secure storage, this is just an example
 api.authenticate(config.API_CLIENT_ID, config.API_CLIENT_SECRET)
 
+
 def get_access_group_CA_key(access_group_id):
     result = api.get_access_group_CA_key(access_group_id)
     if result.ok:
@@ -58,5 +59,6 @@ def main():
     print(ca_id)
     delete_access_group_CA_key(ACCESS_GROUP_ID, ca_id)
 
+
 if __name__ == "__main__":
     main()

examples/get-auditevents.py

@@ -34,7 +34,7 @@
 SEARCH_PARAMETERS = {
     "start_time": "2021-08-01T00:00:00Z",
     "end_time": "2021-10-01T00:00:00Z",
-    "session_id": "607977d7-4791-4ec4-795f-ab0af1ad8eb4"
+    "session_id": "607977d7-4791-4ec4-795f-ab0af1ad8eb4",
 }
 
 

examples/get-connection-tags.py

@@ -50,6 +50,7 @@ def get_connection_tags(offset, limit, sort_dir, query):
     print(connection.data["details"])
     sys.exit(1)
 
+
 def update_connection_tags(conn_id: str, connection_tags):
     """Update connection tags object"""
     result = api.update_connection_tags(conn_id, connection_tags)
@@ -59,11 +60,13 @@ def update_connection_tags(conn_id: str, connection_tags):
     print(result.data["details"])
     sys.exit(1)
 
+
 def main():
     """Update and fetch the connection tags."""
     update_connection_tags(CONNECTION_ID, CONNECTION_TAGS)
     connection_tags = get_connection_tags(OFFSET, LIMIT, SORTDIR, QUERY)
     print(connection_tags)
 
+
 if __name__ == "__main__":
     main()

examples/get-service-status.py

@@ -44,8 +44,9 @@ def main():
     print("Checking for anomalies on PrivX Servers..")
 
     for server_name, microservices in response.data.items():
-        problem_microservices = [m for m in microservices
-                                 if m['status']['status_message'] != 'OK']
+        problem_microservices = [
+            m for m in microservices if m["status"]["status_message"] != "OK"
+        ]
         if problem_microservices:
             print(f"Potential problems with microservice(s) on {server_name}")
             for problem in problem_microservices:

examples/search-connections-trails.py

@@ -53,8 +53,7 @@ def get_connections_data():
     offset = 0
     limit = 1000
     resp = api.search_connections(
-        connection_params={"type": ["SSH"]},
-        offset=offset, limit=limit
+        connection_params={"type": ["SSH"]}, offset=offset, limit=limit
     )
     if resp.ok:
         data_load = resp.data
@@ -63,8 +62,7 @@ def get_connections_data():
         while count > 0:
             offset = offset + limit
             resp = api.search_connections(
-                connection_params={"type": ["SSH"]},
-                offset=offset, limit=limit
+                connection_params={"type": ["SSH"]}, offset=offset, limit=limit
             )
             if resp.ok:
                 data_load = resp.data
@@ -73,7 +71,7 @@ def get_connections_data():
         audited_data_items = []
         for data in data_items:
             if data.get("audit_enabled"):
-                audited_data_items.append(data['id'])
+                audited_data_items.append(data["id"])
         return audited_data_items
     else:
         error = "Get users Connection data operation failed:"
@@ -122,7 +120,7 @@ def process_connections(connections, SEARCH_STRING):
     for connection in connections:
         if "channels" in connection.get("trail", ""):
             for channel in connection["trail"]["channels"]:
-                channel_status = channel["protocol_file"]['status']
+                channel_status = channel["protocol_file"]["status"]
                 if channel["type"] == "shell" and channel_status != "UNCLEAN_CLOSE":
                     # Get a session ID for trail download
                     session = create_trail_session(connection["id"], channel["id"])
@@ -132,8 +130,11 @@ def process_connections(connections, SEARCH_STRING):
                     )
                     # Parse the trail
                     details = (
-                        "\r\nstdin conn " + connection["id"]
-                        + " chan " + channel["id"] + " : "
+                        "\r\nstdin conn "
+                        + connection["id"]
+                        + " chan "
+                        + channel["id"]
+                        + " : "
                     )
                     print_trail(trail_log, SEARCH_STRING, details)
 
@@ -158,13 +159,12 @@ def process_error(messages):
 def main():
     SEARCH_STRING = False
     CONNECTION_ID = False
-    if (len(sys.argv) > 5 or len(sys.argv) == 4 or len(sys.argv) == 1):
+    if len(sys.argv) > 5 or len(sys.argv) == 4 or len(sys.argv) == 1:
         usage()
         sys.exit(2)
     try:
         opts, args = getopt.getopt(
-            sys.argv[1:], "hc:s:",
-            ["help", "connection-id=", "search-string="]
+            sys.argv[1:], "hc:s:", ["help", "connection-id=", "search-string="]
         )
     except getopt.GetoptError:
         usage()

privx_api/authorizer.py

@@ -530,7 +530,7 @@ def delete_access_group(self, access_group_id: str) -> PrivXAPIResponse:
             UrlEnum.AUTHORIZER.ACCESS_GROUP, path_params={"id": access_group_id}
         )
         return PrivXAPIResponse(response_status, HTTPStatus.OK, data)
-    
+
     def get_access_group_CA_key(
         self,
         access_group_id: str,
@@ -546,16 +546,19 @@ def get_access_group_CA_key(
             path_params={"id": access_group_id},
         )
         return PrivXAPIResponse(response_status, HTTPStatus.OK, data)
-    
-    def delete_access_group_CA_key(self, access_group_id: str, ca_id: str) -> PrivXAPIResponse:
+
+    def delete_access_group_CA_key(
+        self, access_group_id: str, ca_id: str
+    ) -> PrivXAPIResponse:
         """
         Delete access group CA key.
 
         Returns:
-            PrivXStreamResponse
+            PrivXAPIResponse
         """
         response_status, data = self._http_delete(
-            UrlEnum.AUTHORIZER.DELETE_ACCESS_GROUP_CA_KEY, path_params={"id": access_group_id, "ca_id": ca_id}
+            UrlEnum.AUTHORIZER.DELETE_ACCESS_GROUP_CA_KEY,
+            path_params={"id": access_group_id, "ca_id": ca_id},
         )
         return PrivXAPIResponse(response_status, HTTPStatus.OK, data)
 

privx_api/connection_manager.py

@@ -113,9 +113,11 @@ def get_connection_tags(
             query_params=search_params,
         )
         return PrivXAPIResponse(response_status, HTTPStatus.OK, data)
-    
+
     def update_connection_tags(
-        self, connection_id: str, connection_tags
+        self,
+        connection_id: str,
+        connection_tags: Optional[list],
     ) -> PrivXAPIResponse:
         """
         Update connection tags

privx_api/enums.py

@@ -203,7 +203,6 @@ class ConnectionManagerEnum:
     CONNECTION_TAGS = "CONNECTION_MANAGER.CONNECTION_TAGS"
     UPDATE_CONNECTION_TAGS = "CONNECTION_MANAGER.UPDATE_CONNECTION_TAGS"
 
-
     urls = {
         ACCESS_ROLE: "/connection-manager/api/v1/connections/access_roles/{role_id}",
         CONNECTION: "/connection-manager/api/v1/connections/{connection_id}",
@@ -240,7 +239,8 @@ class ConnectionManagerEnum:
         UEBA_STATUS: "/connection-manager/api/v1/ueba/status",
         UEBA_INTERNAL_STATUS: "/connection-manager/api/v1/ueba/status/internal",
         CONNECTION_TAGS: "/connection-manager/api/v1/connections/tags",
-        UPDATE_CONNECTION_TAGS: "/connection-manager/api/v1/connections/{connection_id}/tags",
+        UPDATE_CONNECTION_TAGS: "/connection-manager/api/v1/connections/"
+        "{connection_id}/tags",
     }