import demistomock as demisto
from CommonServerPython import * # noqa # pylint: disable=unused-wildcard-import
from CommonServerUserPython import * # noqa
""" IMPORTS """
import urllib
from typing import Any
import urllib3
# Disable insecure warnings
urllib3.disable_warnings()
""" CLIENT CLASS """
class Client(BaseClient):
"""Client class to interact with the service API"""
def __init__(self, username, password, *args, **kwargs):
super().__init__(*args, **kwargs)
self.username = username
self.password = password
def __enter__(self):
self.connect()
return self
def __exit__(self, _type: Any, *args: Any):
self.close()
def connect(self):
"""User login"""
self._http_request(
method="POST",
url_suffix=f"/login?{urllib.parse.urlencode({'userName':self.username,'pass':self.password})}",
resp_type="xml",
)
def close(self):
"""User logout"""
self._http_request(method="POST", url_suffix="/logout", resp_type="text")
def get_lists(self, list_name: str | None = None, list_type: str | None = None) -> str:
"""Gets all available lists using the '/list' API endpoint"""
url_suffix = "/list"
list_filter = []
if list_name:
list_filter.append(f"name={list_name}")
if list_type:
list_filter.append(f"type={list_type}")
if list_filter:
url_suffix += f"?{'&'.join(list_filter)}"
return self._http_request(method="GET", url_suffix=url_suffix, resp_type="text")
def get_list(self, list_id: str):
return self._http_request(method="GET", url_suffix=f"/list/{list_id}", resp_type="text")
def get_list_entry(self, list_id: str, entry_pos: str):
return self._http_request(
method="GET",
url_suffix=f"/list/{list_id}/entry/{entry_pos}",
resp_type="text",
ok_codes=(200, 202, 203, 404),
)
def commit(self):
return self._http_request(method="POST", url_suffix="/commit", resp_type="text")
def put_list(self, list_id: str, config: bytes):
return self._http_request(method="PUT", url_suffix=f"/list/{list_id}", data=config, resp_type="text")
def insert_entry(self, list_id: str, entry_pos: str, data: str):
return self._http_request(
method="POST",
url_suffix=f"/list/{list_id}/entry/{entry_pos}/insert",
data=data,
resp_type="text",
)
def delete_entry(self, list_id: str, entry_pos: str):
return self._http_request(
method="DELETE",
url_suffix=f"/list/{list_id}/entry/{entry_pos}",
resp_type="text",
)
def create_list(self, data: str):
return self._http_request(method="POST", url_suffix="/list", resp_type="text", data=data)
def delete_list(self, list_id: str):
return self._http_request(method="Delete", url_suffix=f"/list/{list_id}", resp_type="text")
""" COMMAND FUNCTIONS """
def test_module(client: Client, args: dict[str, Any]) -> str:
"""Tests API connectivity and authentication'
Returning 'ok' indicates that the integration works like it is supposed to.
Connection to the service is successful.
Raises exceptions if something goes wrong.
Args:
client (Client): API client to use.
args (dict): all command arguments, usually passed from ``demisto.args()``.
Returns:
str: 'ok' if test passed, anything else will raise an exception and will fail the test.
"""
client.get_lists()
return "ok"
def get_lists_command(client: Client, args: dict[str, Any]) -> CommandResults:
"""
get lists command: Returns available lists for matching pattern
Args:
client (Client): API client to use.
args (dict): all command arguments, usually passed from ``demisto.args()``.
``args['name']`` is used to filter lists by name.
``args['type']`` is used to filter lists by type.
Returns:
CommandResults: A ``CommandResults`` object that is then passed to ``return_results``, that contains available lists.
"""
list_name = args.get("name", "")
list_type = args.get("type", "")
res = []
result = client.get_lists(list_name, list_type)
data = json.loads(xml2json(result))
title = demisto.get(data, "feed.title")
entries = demisto.get(data, "feed.entry", [])
if isinstance(entries, dict):
entries = [entries]
for entry in entries:
res.append(
{
"Title": entry.get("title", ""),
"ID": entry.get("id", ""),
"Type": entry.get("listType", ""),
}
)
return CommandResults(
readable_output=tableToMarkdown(title, res, headers=["Title", "ID", "Type"]),
outputs_prefix="SWG.List",
outputs_key_field="ID",
outputs=res,
raw_response=result,
)
def get_list_command(client: Client, args: dict[str, Any]) -> CommandResults | str:
"""
get list command: Returns list details and content
Args:
client (Client): API client to use.
args (dict): all command arguments, usually passed from ``demisto.args()``.
``args['list_id']`` the list id to query.
Returns:
CommandResults: A ``CommandResults`` object that is then passed to ``return_results``,
that contains the list details and content.
"""
resEntries = []
list_id: str = args.get("list_id", "")
result = client.get_list(list_id)
data = json.loads(xml2json(result))
config = json2xml(demisto.get(data, "entry.content", data)).decode("utf-8")
title = demisto.get(data, "entry.title")
res = {
"ID": list_id,
"Title": title,
"Type": demisto.get(data, "entry.listType", ""),
"Description": demisto.get(data, "entry.content.list.description"),
}
entries = demisto.get(data, "entry.content.list.content.listEntry", [])
if isinstance(entries, dict):
entries = [entries]
for pos, entry in enumerate(entries):
description = entry.get("description")
if not description:
description = ""
resEntries.append(
{
"ListID": list_id,
"Position": str(pos),
"Name": entry.get("entry", ""),
"Description": description,
}
)
hr = tableToMarkdown("List Properties", res, headers=["Title", "ID", "Description", "Type"])
res["ListEntries"] = resEntries
return CommandResults(
readable_output=hr + tableToMarkdown(title, resEntries, headers=["Position", "Name", "Description"]),
outputs_prefix="SWG.List",
outputs_key_field=["ID"],
outputs=res,
raw_response=config,
)
def get_list_entry_command(client: Client, args: dict[str, Any]) -> CommandResults | str:
"""
get list entry command: Returns a single entry form a list
Args:
client (Client): API client to use.
args (dict): all command arguments, usually passed from ``demisto.args()``.
``args['list_id']`` the list id to query.
``args['entry_pos']`` the entry position to query.
Returns:
CommandResults: A ``CommandResults`` object that is then passed to ``return_results``, that contains the list entry.
"""
list_id: str = args.get("list_id", "")
entry_pos: str = args.get("entry_pos", "")
result = client.get_list_entry(list_id, entry_pos)
data = json.loads(xml2json(result))
title = demisto.get(data, "entry.title")
entry = demisto.get(data, "entry.content.listEntry", {})
description = entry.get("description")
if not description:
description = ""
res = {
"ID": list_id,
"ListEntries": [
{
"ListID": list_id,
"Position": entry_pos,
"Name": entry.get("entry", ""),
"Description": description,
}
],
}
return CommandResults(
readable_output=tableToMarkdown(title, res["ListEntries"], headers=["ListID", "Position", "Name", "Description"]),
outputs_prefix="SWG.List",
outputs_key_field=["ID"],
outputs=res,
raw_response=result,
)
def modify_list_command(client: Client, args: dict[str, Any]) -> CommandResults | str:
"""
modify list command: Modify the list content
Args:
client (Client): API client to use.
args (dict): all command arguments, usually passed from ``demisto.args()``.
``args['list_id']`` the list id to be modified.
``args['config']`` the config that should be modified to.
Returns:
CommandResults: A ``CommandResults`` object that is then passed to ``return_results``,
that contains the list details and content.
"""
resEntries = []
list_id: str = args.get("list_id", "")
conf: str = args.get("config", "")
result = client.put_list(list_id, conf.encode("utf-8"))
client.commit()
data = json.loads(xml2json(result))
config = json2xml(demisto.get(data, "entry.content", data)).decode("utf-8")
title = demisto.get(data, "entry.title", "")
description = demisto.get(data, "entry.content.list.description")
if not description:
description = ""
res = {
"ID": list_id,
"Title": title,
"Type": demisto.get(data, "entry.listType", ""),
"Description": description,
}
entries = demisto.get(data, "entry.content.list.content.listEntry", [])
if isinstance(entries, dict):
entries = [entries]
for pos, entry in enumerate(entries):
description = entry.get("description")
if not description:
description = ""
resEntries.append(
{
"ListID": list_id,
"Position": str(pos),
"Name": entry.get("entry", ""),
"Description": description,
}
)
hr = tableToMarkdown("List Properties", res, headers=["Title", "ID", "Description", "Type"])
res["ListEntries"] = resEntries
return CommandResults(
readable_output=hr + tableToMarkdown(title, resEntries, headers=["Position", "Name", "Description"]),
outputs_prefix="SWG.List",
outputs_key_field=["ID"],
outputs=res,
raw_response=config,
)
def insert_entry_command(client: Client, args: dict[str, Any]) -> CommandResults | str:
"""
insert entry command: Insert an entry to the list
Args:
client (Client): API client to use.
args (dict): all command arguments, usually passed from ``demisto.args()``.
``args['list_id']`` the list id to be modified.
``args['entry_pos']`` the entry position to insert at.
Returns:
CommandResults: A ``CommandResults`` object that is then passed to ``return_results``,
that contains the inserted list entry.
"""
list_id: str = args.get("list_id", "")
entry_pos: str = args.get("entry_pos", "")
name: str = args.get("name", "")
description: str = args.get("description", "")
entry = f"{name}{description}"
result = client.insert_entry(list_id, entry_pos, entry)
client.commit()
data = json.loads(xml2json(result))
title = f'Added {demisto.get(data, "entry.title")}'
entry = demisto.get(data, "entry.content.listEntry", {})
description = entry.get("description")
if not description:
description = ""
entry_name = entry.get("entry", "")
list_entry = {
"ListID": list_id,
"Position": entry_pos,
"Name": entry_name,
"Description": description,
}
swg_lists = demisto.get(demisto.context(), "SWG.List", [])
if isinstance(swg_lists, dict):
swg_lists = [swg_lists]
if any(item["ID"] == list_id for item in swg_lists):
outputs_prefix = f'SWG.List(val.ID && val.ID == "{list_id}").ListEntries'
outputs_key_field = "Name"
res = list_entry
else:
outputs_prefix = "SWG.List"
outputs_key_field = "ID"
res = {"ID": list_id, "ListEntries": [list_entry]}
return CommandResults(
readable_output=tableToMarkdown(title, list_entry, headers=["ListID", "Position", "Name", "Description"]),
outputs_prefix=outputs_prefix,
outputs_key_field=outputs_key_field,
outputs=res,
raw_response=result,
)
def delete_entry_command(client: Client, args: dict[str, Any]) -> CommandResults | str:
"""
delete entry command: Delete the list entry
Args:
client (Client): API client to use.
args (dict): all command arguments, usually passed from ``demisto.args()``.
``args['list_id']`` the list id to be modified.
``args['entry_pos']`` the entry position to be deleted.
Returns:
CommandResults: A ``CommandResults`` object that is then passed to ``return_results``, that contains the list entry.
"""
list_id: str = args.get("list_id", "")
entry_pos: str = args.get("entry_pos", "")
result = client.delete_entry(list_id, entry_pos)
client.commit()
data = json.loads(xml2json(result))
title = f'Deleted {demisto.get(data, "entry.title")}'
entry = demisto.get(data, "entry.content.listEntry", {})
description = entry.get("description")
if not description:
description = ""
res = {
"ID": list_id,
"ListEntries": [
{
"ListID": list_id,
"Position": entry_pos,
"Name": entry.get("entry", ""),
"Description": description,
}
],
}
return CommandResults(
readable_output=tableToMarkdown(title, res["ListEntries"], headers=["ListID", "Position", "Name", "Description"]),
raw_response=result,
)
def create_list_command(client: Client, args: dict[str, Any]) -> CommandResults | str:
"""
create list command: Create an empty list
Args:
client (Client): API client to use.
args (dict): all command arguments, usually passed from ``demisto.args()``.
``args['name']`` the list name to be added.
``args['type']`` the list type to be added.
Returns:
CommandResults: A ``CommandResults`` object that is then passed to ``return_results``, that contains the list entry.
"""
list_name: str = args.get("name", "")
list_type: str = args.get("type", "")
list_data = (
f'
'
)
result = client.create_list(list_data)
client.commit()
data = json.loads(xml2json(result))
config = json2xml(demisto.get(data, "entry.content", data)).decode("utf-8")
title = demisto.get(data, "entry.title")
description = demisto.get(data, "entry.content.list.description")
if not description:
description = ""
res = {
"ID": demisto.get(data, "entry.id"),
"Title": title,
"Type": demisto.get(data, "entry.listType"),
"Description": description,
}
hr = tableToMarkdown("Created List Properties", res, headers=["Title", "ID", "Description", "Type"])
return CommandResults(
readable_output=hr, outputs_prefix="SWG.List", outputs_key_field=["ID"], outputs=res, raw_response=config
)
def delete_list_command(client: Client, args: dict[str, Any]) -> CommandResults | str:
"""
create list command: Create an empty list
Args:
client (Client): API client to use.
args (dict): all command arguments, usually passed from ``demisto.args()``.
``args['list_id']`` the list id to be deleted.
Returns:
CommandResults: A ``CommandResults`` object that is then passed to ``return_results``, that contains the list entry.
"""
list_id: str = args.get("list_id", "")
result = client.delete_list(list_id)
client.commit()
data = json.loads(xml2json(result))
title = demisto.get(data, "entry.title")
description = demisto.get(data, "entry.content.list.description")
if not description:
description = ""
res = {
"ID": demisto.get(data, "entry.id"),
"Title": title,
"Type": demisto.get(data, "entry.listType"),
"Description": description,
}
hr = tableToMarkdown("Deleted List Properties", res, headers=["Title", "ID", "Description", "Type"])
return CommandResults(readable_output=hr, raw_response=result)
""" MAIN FUNCTION """
def main() -> None:
"""
main function, parses params and runs command functions
"""
user = demisto.params().get("credentials", {}).get("identifier")
password = demisto.params().get("credentials", {}).get("password")
base_url = urljoin(demisto.params()["url"], "/Konfigurator/REST")
verify_certificate = not demisto.params().get("insecure", False)
proxy = demisto.params().get("proxy", False)
headers = {"Content-Type": "application/mwg+xml"}
command = demisto.command()
demisto.debug(f"Command being called is {command}")
try:
with Client(
username=user,
password=password,
base_url=base_url,
verify=verify_certificate,
headers=headers,
proxy=proxy,
) as client:
commands = {
"test-module": test_module,
"swg-get-available-lists": get_lists_command,
"swg-get-list": get_list_command,
"swg-get-list-entry": get_list_entry_command,
"swg-modify-list": modify_list_command,
"swg-insert-entry": insert_entry_command,
"swg-delete-entry": delete_entry_command,
"swg-create-list": create_list_command,
"swg-delete-list": delete_list_command,
}
if command not in commands:
raise NotImplementedError(f"Command {command} was not implemented.")
return_results(commands[command](client, demisto.args()))
except Exception as e:
return_error(f"Failed to execute {command} command.\nError:\n{e!s}")
""" ENTRY POINT """
if __name__ in ("__main__", "__builtin__", "builtins"):
main()