MicrosoftGraphFilesApiModule
Common Microsoft Graph Files code that will be appended into the Microsoft Graph Files integrations when it's deployed.
- Type
- python
- Pack
- ApiModules
Source
from CommonServerPython import * # noqa: F401
import demistomock as demisto # noqa: F401
""" IMPORTS """
import base64
from urllib.parse import parse_qs, quote, urlparse
from MicrosoftApiModule import * # noqa: E402
""" GLOBALS/PARAMS """
INTEGRATION_NAME = "MsGraphFiles"
APP_NAME = "ms-graph-files"
RESPONSE_KEYS_DICTIONARY = {
"@odata.context": "OdataContext",
"@microsoft.graph.downloadUrl": "DownloadUrl",
"id": "ID",
"@odata.nextLink": "OdataNextLink",
}
EXCLUDE_LIST = ["eTag", "cTag", "quota"]
# The MS Graph resources a driveItem can be addressed under.
VALID_OBJECT_TYPES = ("drives", "groups", "sites", "users")
# Listed explicitly because $select narrows the projection - anything omitted here will
# be absent from the response.
DRIVEITEM_SELECT_FIELDS = (
"id,name,size,webUrl,createdDateTime,lastModifiedDateTime,createdBy,lastModifiedBy,parentReference,file,folder"
)
# The role keys an identitySet may nest an identity under, in preference order. 'siteUser'
# matters for SharePoint sharing entries, where it is often the only one populated.
IDENTITY_ROLE_KEYS = ("user", "siteUser", "group", "application", "device")
def parse_key_to_context(obj: dict) -> dict:
"""Parse graph api data as received from Microsoft Graph API into Demisto's conventions
Args:
item object: a dictionary containing the item data
Returns:
A Camel Cased dictionary with the relevant fields.
groups_readable: for the human readable
groups_outputs: for the entry context
"""
parsed_obj = {}
for key, value in obj.items():
if key in EXCLUDE_LIST:
continue
new_key = RESPONSE_KEYS_DICTIONARY.get(key, key)
parsed_obj[new_key] = value
if isinstance(value, dict):
parsed_obj[new_key] = parse_key_to_context(value)
under_score_obj = createContext(parsed_obj, keyTransform=camel_case_to_underscore)
context_entry: dict = createContext(under_score_obj, keyTransform=string_to_context_key)
if "Id" in list(context_entry.keys()):
context_entry["ID"] = context_entry["Id"]
del context_entry["Id"]
if "CreatedBy" in list(context_entry.keys()):
context_entry["CreatedBy"] = remove_identity_key(context_entry["CreatedBy"])
if "LastModifiedBy" in list(context_entry.keys()):
context_entry["LastModifiedBy"] = remove_identity_key(context_entry["LastModifiedBy"])
return context_entry
def remove_identity_key(source: Any) -> dict:
"""
this function removes identity key (application, device or user) from LastModifiedBy and CreatedBy keys and
convert it to "type" key.
:param source: LastModifiedBy and CreatedBy dictionaries
:return: camel case dictionary with identity key as type.
"""
if not isinstance(source, dict):
demisto.debug("Input is not dictionary. Exist function.")
return source
dict_keys = list(source.keys())
if len(dict_keys) != 1:
demisto.debug("Got more then one identity creator. Exit function")
return source
identity_key = dict_keys[0]
new_source = {}
if source[identity_key].get("ID"):
new_source["ID"] = source[identity_key].get("ID")
new_source["DisplayName"] = source[identity_key].get("DisplayName")
new_source["Type"] = identity_key
return new_source
def encode_sharing_url(share_url: str) -> str:
"""Encode a sharing URL into the token accepted by GET /shares/{token}.
Per the shares API: base64url-encode, strip the '=' padding, prefix 'u!'.
The Microsoft sample spells this out as base64 followed by two explicit replacements
('/' -> '_' and '+' -> '-'). urlsafe_b64encode already emits that alphabet, so the
replacements are performed here, not missing. See "Encoding sharing URLs":
https://learn.microsoft.com/en-us/graph/api/shares-get
"""
encoded = base64.urlsafe_b64encode(share_url.encode("utf-8")).decode("utf-8")
return f"u!{encoded.rstrip('=')}"
def validate_object_type(object_type: str) -> None:
"""Raise if object_type is not a resource a driveItem can be addressed under."""
if object_type not in VALID_OBJECT_TYPES:
raise DemistoException(f"Invalid object_type '{object_type}'. Must be one of: {', '.join(VALID_OBJECT_TYPES)}.")
def _select_addressing_mode(addressing_args: dict[str, str], allow_path: bool, allow_share_url: bool) -> str:
"""Return the name of the single addressing argument supplied, or raise explaining why not.
Raises:
DemistoException: If an unsupported argument is used, or the number supplied is not one.
"""
if addressing_args["item_path"] and not allow_path:
raise DemistoException(
"The item_path argument is not supported by this command. Microsoft Graph does not "
"document path addressing for this endpoint. Use item_id instead."
)
if addressing_args["share_url"] and not allow_share_url:
raise DemistoException("The share_url argument is not supported by this command. Use item_id instead.")
supplied = [name for name, value in addressing_args.items() if value]
if len(supplied) == 1:
return supplied[0]
allowed = ["item_id"] + (["item_path"] if allow_path else []) + (["share_url"] if allow_share_url else [])
if not supplied:
raise DemistoException(f"Provide one of the following arguments: {', '.join(allowed)}.")
raise DemistoException(f"Provide only one of the following arguments, but got {', '.join(supplied)}.")
def resolve_item_addressing(args: dict[str, str], allow_path: bool = True, allow_share_url: bool = False) -> dict[str, str]:
"""Validate and normalize the arguments that address a single driveItem.
Graph offers several ways to address the same item, and which are valid differs per
endpoint. YAML cannot express "exactly one of these", so the rule is enforced here.
Returns:
{'mode': 'item_id'|'item_path'|'share_url', 'value': ..., 'object_type': ...,
'object_type_id': ...}. The shape is the same for every mode so callers can index
directly. share_url is self-addressing, so its object type fields are empty.
Raises:
DemistoException: If the addressing arguments are missing, ambiguous or unsupported.
"""
# Insertion order drives the order names appear in the error messages.
addressing_args = {name: args.get(name) or "" for name in ("item_id", "item_path", "share_url")}
mode = _select_addressing_mode(addressing_args, allow_path, allow_share_url)
if mode == "share_url":
# A sharing URL resolves via /shares/{token}, which needs no parent resource.
return {"mode": "share_url", "value": addressing_args["share_url"], "object_type": "", "object_type_id": ""}
object_type_id = args.get("object_type_id") or ""
if not object_type_id:
raise DemistoException(f"The object_type_id argument is required when addressing an item by {mode}.")
# object_type has no default, so an omitted value would otherwise build a malformed URL
# such as '/{object_type_id}/drive/items/{item_id}'. Fail here with a usable message.
object_type = args.get("object_type") or ""
if object_type not in VALID_OBJECT_TYPES:
raise DemistoException(
f"The object_type argument is required when addressing an item by {mode}, "
f"and must be one of: {', '.join(VALID_OBJECT_TYPES)}. Got '{object_type}'."
)
return {
"mode": mode,
"value": addressing_args[mode],
"object_type": object_type,
"object_type_id": object_type_id,
}
def url_validation(url: str) -> str:
"""
this function tests if a user provided a valid next link url
:param url: next_link_url from graph api
:return: checked url if url is valid. demisto error if not.
"""
parsed_url = urlparse(url)
# test if exits $skiptoken
url_parameters = parse_qs(parsed_url.query)
if not url_parameters.get("$skiptoken") or not url_parameters["$skiptoken"]:
raise DemistoException(f"Url: {url} is not valid. Please provide another one. missing $skiptoken")
return url
class MsGraphClient:
"""
Microsoft Graph Client enables authorized access to organization's files in OneDrive, SharePoint, and MS Teams.
"""
MAX_ATTACHMENT_SIZE = 3145728 # 3mb = 3145728 bytes
MAX_ATTACHMENT_UPLOAD = 327680 # 320 KiB = 327680 bytes
def __init__(
self,
tenant_id,
auth_id,
enc_key,
app_name,
base_url,
verify,
proxy,
self_deployed,
ok_codes,
redirect_uri,
auth_code,
certificate_thumbprint: Optional[str] = None,
private_key: Optional[str] = None,
managed_identities_client_id: Optional[str] = None,
):
# Under UCP (ConnectUs) the auth secrets are not supplied via
# demisto.params() - they are injected per-request by the platform
if not managed_identities_client_id and not should_use_ucp_auth():
if not self_deployed and not enc_key:
raise DemistoException(
"Key must be provided. For further information see "
"https://xsoar.pan.dev/docs/reference/articles/microsoft-integrations---authentication"
)
if self_deployed and (not enc_key and not (certificate_thumbprint and private_key)):
raise DemistoException(
"Either Key or (Certificate Thumbprint and Private Key) must be provided. For further "
"information see "
"https://xsoar.pan.dev/docs/reference/articles/microsoft-integrations---authentication"
)
elif self_deployed and auth_code and not redirect_uri:
raise DemistoException(
"Please provide both Application redirect URI and Authorization code "
"for Authorization Code flow, or None for the Client Credentials flow"
)
grant_type = AUTHORIZATION_CODE if auth_code and redirect_uri else CLIENT_CREDENTIALS
self.ms_client = MicrosoftClient(
tenant_id=tenant_id,
auth_id=auth_id,
enc_key=enc_key,
app_name=app_name,
base_url=base_url,
verify=verify,
proxy=proxy,
self_deployed=self_deployed,
ok_codes=ok_codes,
certificate_thumbprint=certificate_thumbprint,
private_key=private_key,
managed_identities_client_id=managed_identities_client_id,
managed_identities_resource_uri=Resources.graph,
redirect_uri=redirect_uri,
auth_code=auth_code,
grant_type=grant_type,
command_prefix="msgraph-files",
)
def list_sharepoint_sites(self, keyword: str) -> dict:
"""
This function lists the tenant sites
:return: graph api raw response
"""
return self.ms_client.http_request(
method="GET",
url_suffix="sites",
params={"search": keyword},
)
def list_drives_in_site(self, site_id=None, limit=None, next_page_url=None) -> dict:
"""
Returns the list of Drive resources available for a target Site
:param site_id: selected Site ID.
:param limit: sets the page size of results.
:param next_page_url: the URL for the next results page.
:return:
"""
if not any([site_id, next_page_url]):
raise DemistoException(
"Please pass at least one argument to this command: \n"
"site_id: if you want to get all sites.\n"
"next_page_url: if you have used the limit argument."
)
params = {"$top": limit} if limit else ""
if next_page_url:
url = url_validation(next_page_url)
return self.ms_client.http_request(method="GET", full_url=url, params=params)
url_suffix = f"sites/{site_id}/drives"
return self.ms_client.http_request(method="GET", params=params, url_suffix=url_suffix)
def list_site_permissions(self, site_id: str, permission_id: str | None) -> dict:
"""Lists permissions for a SharePoint site.
Args:
site_id: The ID of the site to list permissions for.
permission_id: The unique identifier for the permission to retrieve.
When not provided, a list of all permissions is returned.
Returns:
dict: The raw response from the Graph API.
Note:
When the permission_id parameter is not provided in the command,
a list of permissions is returned, but the `roles` field is not included in each permission object.
To get the `roles` field, you must provide the permission_id parameter.
"""
url_suffix = f"sites/{site_id}/permissions"
if permission_id:
url_suffix += f"/{permission_id}"
return self.ms_client.http_request(method="GET", url_suffix=url_suffix)
def create_site_permission(self, site_id: str, app_id: str, display_name: str, role: list[str]) -> dict:
url_suffix = f"sites/{site_id}/permissions"
body = {"roles": role, "grantedToIdentities": [{"application": {"id": app_id, "displayName": display_name}}]}
return self.ms_client.http_request(method="POST", url_suffix=url_suffix, json_data=body)
def update_site_permission(self, site_id: str, permission_id: str, role: list[str]) -> dict:
url_suffix = f"sites/{site_id}/permissions/{permission_id}"
body = {"roles": role}
return self.ms_client.http_request(method="PATCH", url_suffix=url_suffix, json_data=body)
def delete_site_permission(self, site_id: str, permission_id: str) -> requests.Response:
"""
We will always receive a 204 status code when attempting to remove a permission,
even if the specified permission ID does not exist in the permission list.
"""
url_suffix = f"/sites/{site_id}/permissions/{permission_id}"
return self.ms_client.http_request(method="DELETE", url_suffix=url_suffix, return_empty_response=True)
def list_drive_content(self, object_type: str, object_type_id: str, item_id: str, limit=None, next_page_url=None) -> dict:
"""
This command list all the drive's files and folders
:param object_type: ms graph resource.
:param object_type_id: ms graph resource id.
:param item_id: ms graph item_id. optional.
:param limit: sets the page size of results. optional.
:param next_page_url: the URL for the next results page. optional.
:return: graph api raw response
"""
params = {"$top": limit} if limit else ""
uri = ""
if next_page_url:
url = url_validation(next_page_url)
return self.ms_client.http_request(method="GET", full_url=url, params=params)
if object_type == "drives":
uri = f"{object_type}/{object_type_id}/items/{item_id}/children"
elif object_type in {"groups", "sites", "users"}:
uri = f"{object_type}/{object_type_id}/drive/items/{item_id}/children"
return self.ms_client.http_request(
method="GET",
url_suffix=uri,
params=params,
)
def replace_existing_file(self, object_type, object_type_id, item_id, entry_id):
"""
replace file context in MS Graph resource
:param object_type: ms graph resource.
:param object_type_id: ms graph resource id.
:param item_id: item_id: ms graph item_id.
:param entry_id: demisto file entry id
:return: graph api raw response
"""
uri = ""
file_path = demisto.getFilePath(entry_id).get("path", None)
if not file_path:
raise DemistoException(f"Could not find file path to the next entry id: {entry_id}. \nPlease provide another one.")
if object_type == "drives":
uri = f"{object_type}/{object_type_id}/items/{item_id}/content"
elif object_type in ["groups", "sites", "users"]:
uri = f"{object_type}/{object_type_id}/drive/items/{item_id}/content"
with open(file_path, "rb") as file:
headers = {"Content-Type": "application/octet-stream"}
return self.ms_client.http_request(method="PUT", data=file, headers=headers, url_suffix=uri)
def replace_existing_file_with_upload_session(
self, object_type: str, object_type_id: str, item_id: str, entry_id: str, file_data: bytes, file_size: int, file_name: str
) -> requests.Response:
"""
Replace a file with upload session.
Args:
object_type: ms graph resource.
object_type_id: ms graph resource id.
item_id: ms graph item_id.
entry_id: demisto file entry id
Returns:
MsGraph api raw response.
"""
file_path = demisto.getFilePath(entry_id).get("path", None)
if not file_path:
raise DemistoException(f"Could not find file path to the next entry id: {entry_id}. \nPlease provide another one.")
uri = ""
# create suitable upload session
if object_type == "drives":
uri = f"/drives/{object_type_id}/items/{item_id}/createUploadSession"
elif object_type == "groups":
uri = f"/groups/{object_type_id}/drive/items/{item_id}/createUploadSession"
elif object_type == "sites":
uri = f"/sites/{object_type_id}/drive/items/{item_id}/createUploadSession"
elif object_type == "users":
uri = f"/users/{object_type_id}/drive/items/{item_id}/createUploadSession"
response, upload_url = self.create_an_upload_session(uri)
if not upload_url:
raise Exception(f"Cannot get upload URL for attachment {file_name}")
demisto.debug(f'response of "create_an_upload_session": {response}')
response_file_upload = self.upload_file_with_upload_session(upload_url, file_data, file_size)
demisto.debug(f'response of "upload_file_with_upload_session": {response_file_upload}')
return response_file_upload
@staticmethod
def _driveitem_uri(object_type: str, object_type_id: str, item_id: str, suffix: str = "") -> str:
"""Build a /v1.0 relative URL for a driveItem sub-resource.
Args:
object_type: One of drives, groups, sites, users.
object_type_id: ID of the parent resource.
item_id: ID of the target driveItem.
suffix: Optional path suffix appended after items/{item_id} (no leading slash).
Returns:
Relative URL suitable for ms_client.http_request(url_suffix=...).
"""
base = (
f"{object_type}/{object_type_id}/items/{item_id}"
if object_type == "drives"
else f"{object_type}/{object_type_id}/drive/items/{item_id}"
)
return f"{base}/{suffix}" if suffix else base
@staticmethod
def _driveitem_path_uri(object_type: str, object_type_id: str, item_path: str, suffix: str = "") -> str:
"""Build a /v1.0 relative URL addressing a driveItem by its path under the drive root.
Graph addresses items by path as 'root:/{item-path}', closing with a colon when a
sub-resource follows: 'root:/{item-path}:/{suffix}'. Only some endpoints support this.
"""
# Encode the segments but keep the '/' separators, so spaces and other reserved
# characters in file names do not break the URL.
encoded_path = quote(item_path.strip("/"), safe="/")
prefix = f"{object_type}/{object_type_id}" if object_type == "drives" else f"{object_type}/{object_type_id}/drive"
root = f"{prefix}/root:"
return f"{root}/{encoded_path}:/{suffix}" if suffix else f"{root}/{encoded_path}"
def update_driveitem(
self,
object_type: str,
object_type_id: str,
item_id: str,
body: dict,
) -> dict:
"""Apply a PATCH on a driveItem (move / rename / metadata update).
Args:
object_type: One of drives, groups, sites, users.
object_type_id: ID of the parent resource.
item_id: ID of the driveItem to update.
body: JSON body to send. Only keys the caller provided are included.
Returns:
The updated driveItem object returned by Microsoft Graph.
"""
url_suffix = self._driveitem_uri(object_type, object_type_id, item_id)
return self.ms_client.http_request(method="PATCH", url_suffix=url_suffix, json_data=body)
def list_driveitem_permissions(
self,
object_type: str,
object_type_id: str,
item_id: str,
limit: str | None = None,
next_page_url: str | None = None,
) -> dict:
"""List permissions (sharing entries) on a driveItem.
Args:
object_type: One of drives, groups, sites, users.
object_type_id: ID of the parent resource.
item_id: ID of the driveItem.
limit: Optional $top page size.
next_page_url: Optional full URL from a previous @odata.nextLink to fetch the next page.
Returns:
The raw response from Microsoft Graph (with value[] and optional @odata.nextLink).
"""
params = {"$top": limit} if limit else {}
if next_page_url:
url = url_validation(next_page_url)
return self.ms_client.http_request(method="GET", full_url=url, params=params)
url_suffix = self._driveitem_uri(object_type, object_type_id, item_id, suffix="permissions")
return self.ms_client.http_request(method="GET", url_suffix=url_suffix, params=params)
def delete_driveitem_permission(
self,
object_type: str,
object_type_id: str,
item_id: str,
permission_id: str,
) -> requests.Response:
"""Delete (revoke) a single sharing permission on a driveItem.
Args:
object_type: One of drives, groups, sites, users.
object_type_id: ID of the parent resource.
item_id: ID of the driveItem.
permission_id: ID of the permission to delete (obtain via list_driveitem_permissions).
Returns:
The raw requests.Response from Microsoft Graph (status 204 with no body).
"""
url_suffix = self._driveitem_uri(object_type, object_type_id, item_id, suffix=f"permissions/{permission_id}")
return self.ms_client.http_request(
method="DELETE",
url_suffix=url_suffix,
resp_type="text",
)
def copy_driveitem(
self,
object_type: str,
object_type_id: str,
item_id: str,
body: dict,
params: dict,
) -> requests.Response:
"""Initiate a driveItem copy operation. Returns the raw response so the Location header can be read.
Microsoft Graph responds 202 Accepted with a Location header pointing to a monitor URL
that the caller polls until the copy reaches a terminal status.
Args:
object_type: One of drives, groups, sites, users.
object_type_id: ID of the parent resource.
item_id: ID of the source driveItem.
body: JSON body to send. Only keys the caller provided are included.
params: Query parameters to send. Only keys the caller provided are included.
Returns:
The raw requests.Response from Microsoft Graph (status 202 with a Location header).
"""
url_suffix = self._driveitem_uri(object_type, object_type_id, item_id, suffix="copy")
return self.ms_client.http_request(
method="POST",
url_suffix=url_suffix,
json_data=body,
params=params,
resp_type="response",
)
def delete_file(self, object_type: str, object_type_id: str, item_id: str) -> str:
"""
Delete a DriveItem by using its ID
:param object_type: ms graph resource.
:param object_type_id: ms graph resource id.
:param item_id: ms graph item_id.
:return: graph api raw response
"""
uri = ""
if object_type == "drives":
uri = f"{object_type}/{object_type_id}/items/{item_id}"
elif object_type in {"groups", "sites", "users"}:
uri = f"{object_type}/{object_type_id}/drive/items/{item_id}"
# send request
self.ms_client.http_request(method="DELETE", url_suffix=uri, resp_type="text")
return "Item was deleted successfully"
@staticmethod
def upload_attachment(upload_url, start_chunk_idx, end_chunk_idx, chunk_data, attachment_size) -> requests.Response:
"""
Upload an attachment to the upload URL.
Args:
upload_url (str): upload URL provided when running 'get_upload_session'
start_chunk_idx (int): the start of the chunk file data.
end_chunk_idx (int): the end of the chunk file data.
chunk_data (bytes): the chunk data in bytes from start_chunk_idx to end_chunk_idx
attachment_size (int): the entire attachment size in bytes.
Returns:
Response: response indicating whether the operation succeeded. 200 if a chunk was added successfully,
201 (created) if the file was uploaded completely. 400 in case of errors.
"""
chunk_size = len(chunk_data)
headers = {
"Content-Length": f"{chunk_size}",
"Content-Range": f"bytes {start_chunk_idx}-{end_chunk_idx - 1}/{attachment_size}",
"Content-Type": "application/octet-stream",
}
try:
response = requests.put(url=upload_url, data=chunk_data, headers=headers)
except Exception as e:
raise (e)
return response
def upload_file_with_upload_session(self, upload_url: str, file_data: bytes, file_size: int) -> requests.Response:
"""
Add an attachment using an upload session by dividing the file bytes into chunks and sent each chunk each time.
more info here -
https://learn.microsoft.com/en-us/onedrive/developer/rest-api/api/driveitem_createuploadsession?view=odsp-graph-online#upload-bytes-to-the-upload-session
Args:
upload_url (str): url to file upload.
file_data (bytes): The file data.
file_size (int): The file size in bytes.
Returns:
Response: response indicating whether the operation succeeded. 200 or
201 (created) if the file was uploaded completely. 400 in case of errors.
"""
start_chunk_index = 0
end_chunk_index = self.MAX_ATTACHMENT_UPLOAD
chunk_data = file_data[start_chunk_index:end_chunk_index]
response = self.upload_attachment(
upload_url=upload_url,
start_chunk_idx=start_chunk_index,
end_chunk_idx=end_chunk_index,
chunk_data=chunk_data,
attachment_size=file_size,
)
demisto.debug(f"start_chunk_idx:{start_chunk_index}, end_chunk_idx:{end_chunk_index}")
while response.status_code not in [201, 200]: # the api returns 201 when the file is created
start_chunk_index = end_chunk_index
next_chunk = end_chunk_index + self.MAX_ATTACHMENT_UPLOAD
end_chunk_index = min(next_chunk, file_size)
chunk_data = file_data[start_chunk_index:end_chunk_index]
demisto.debug(f"start_chunk_idx:{start_chunk_index}, end_chunk_idx:{end_chunk_index}")
response = self.upload_attachment(
upload_url=upload_url,
start_chunk_idx=start_chunk_index,
end_chunk_idx=end_chunk_index,
chunk_data=chunk_data,
attachment_size=file_size,
)
if response.status_code not in (201, 200, 202):
raise Exception(f"{response.json()}")
return response
def create_an_upload_session(self, uri: str) -> tuple:
"""
Creates an upload session to the file.
Args:
uri (str): uri of the request.
file_name (str): the name of the file.
Returns:
Response: The response to this request, if successful, will provide the details for where the
remainder of the requests should be sent as an UploadSession resource.
Upload_url: A url upload resource.
"""
request_body = {"item": {"@microsoft.graph.conflictBehavior": "replace"}}
response = self.ms_client.http_request(method="POST", json_data=request_body, url_suffix=uri)
return response, response.get("uploadUrl")
def upload_file_with_upload_session_flow(
self, object_type: str, object_type_id: str, parent_id: str, file_name: str, file_data: bytes, file_size: int
) -> requests.Response:
"""
Uploads a file with the upload session flow, this is used only when the file is larger
than 3 MB.
Args:
object_type (str): drive/ group/ site/ users
object_type_id (str): the selected object type id.
parent_id (str): an ID of the folder to upload the file to.
file_name (str): file name.
file_data (bytes): The file data.
file_size (int): The file size in bytes.
Returns:
Response: response indicating whether the operation succeeded. 200 or
201 (created) if the file was uploaded completely. 400 in case of errors.
"""
# create suitable upload session
uri = ""
if object_type == "drives":
uri = f"/drives/{object_type_id}/items/{parent_id}:/{file_name}:/createUploadSession"
elif object_type == "groups":
uri = f"/groups/{object_type_id}/drive/items/{parent_id}:/{file_name}:/createUploadSession"
elif object_type == "sites":
uri = f"/sites/{object_type_id}/drive/items/{parent_id}:/{file_name}:/createUploadSession"
elif object_type == "users":
uri = f"/users/{object_type_id}/drive/items/{parent_id}:/{file_name}:/createUploadSession"
response, upload_url = self.create_an_upload_session(uri)
if not upload_url:
raise Exception(f"Cannot get upload URL for attachment {file_name}")
demisto.debug(f'Create upload session response": {response}')
response_file_upload = self.upload_file_with_upload_session(upload_url, file_data, file_size)
demisto.debug(f'response of "upload_file_with_upload_session": {response}')
return response_file_upload
def upload_new_file(self, object_type: str, object_type_id: str, parent_id: str, file_name: str, entry_id: str):
"""
this function upload new file to a selected folder(parent_id)
:param object_type: drive/ group/ site/ users
:param object_type_id: the selected object type id.
:param parent_id: an ID of the folder to upload the file to.
:param file_name: file name
:param entry_id: demisto file entry ID.
:return: graph api raw response.
"""
file_path: str = demisto.getFilePath(entry_id).get("path", "")
uri = ""
if object_type == "drives":
uri = f"{object_type}/{object_type_id}/items/{parent_id}:/{file_name}:/content"
elif object_type in {"groups", "users", "sites"}:
uri = f"{object_type}/{object_type_id}/drive/items/{parent_id}:/{file_name}:/content"
with open(file_path, "rb") as file:
headers = {"Content-Type": "application/octet-stream"}
return self.ms_client.http_request(method="PUT", headers=headers, url_suffix=uri, data=file)
def download_file(self, object_type: str, object_type_id: str, item_id: str) -> requests.Response:
"""
Download the contents of the file of a DriveItem.
:param object_type: ms graph resource.
:param object_type_id: the selected object type id.
:param item_id: ms graph item_id.
:return: graph api raw response
"""
uri = ""
if object_type == "drives":
uri = f"{object_type}/{object_type_id}/items/{item_id}/content"
elif object_type in {"groups", "sites", "users"}:
uri = f"{object_type}/{object_type_id}/drive/items/{item_id}/content"
return self.ms_client.http_request(method="GET", url_suffix=uri, resp_type="response")
def create_new_folder(self, object_type: str, object_type_id: str, parent_id: str, folder_name: str) -> dict:
"""
Create a new folder in a Drive with a specified parent item or path.
:param object_type: ms graph resource.
:param object_type_id: the selected object type id.
:param parent_id: an ID of the parent to upload the folder to.
:param folder_name: folder name
:return: graph api raw response
"""
uri = ""
if object_type == "drives":
uri = f"{object_type}/{object_type_id}/items/{parent_id}/children"
elif object_type in {"groups", "sites", "users"}:
uri = f"{object_type}/{object_type_id}/drive/items/{parent_id}/children"
# send request
payload = {
"name": folder_name,
"folder": {},
"@microsoft.graph.conflictBehavior": "rename",
}
return self.ms_client.http_request(method="POST", json_data=payload, url_suffix=uri)
@staticmethod
def _build_drive_item_uri(object_type: str, object_type_id: str, item_id: str) -> str:
"""Build the drive-item URI prefix based on the MS Graph resource type.
Args:
object_type: MS Graph resource. One of 'drives', 'groups', 'sites', 'users'.
object_type_id: MS Graph resource ID.
item_id: The drive item ID.
Returns:
Relative URI of the drive item, with no trailing slash.
"""
if object_type == "drives":
return f"drives/{object_type_id}/items/{item_id}"
if object_type in {"groups", "sites", "users"}:
return f"{object_type}/{object_type_id}/drive/items/{item_id}"
raise DemistoException(f"Invalid object_type '{object_type}'. Must be one of: drives, groups, sites, users.")
def get_sharepoint_ids(self, site_id: str, item_id: str, drive_id: str = "") -> dict:
"""Retrieve the SharePoint identifiers of a driveItem, or {} if it exposes none.
sharepointIds is not returned by a regular driveItem request, so it has to be asked
for on its own. It carries listId and listItemUniqueId, the join keys between the
drive world and the SharePoint list world.
Addressing is by drive whenever the drive ID is known, because
sites/{site-id}/drive resolves only the site's *default* document library - an item
in any other library, or in a user's OneDrive (what a sharing URL usually points at),
is not reachable that way. The site route is kept for callers that have no drive ID.
"""
url_suffix = f"drives/{drive_id}/items/{item_id}" if drive_id else f"sites/{site_id}/drive/items/{item_id}"
response = self.ms_client.http_request(
method="GET",
url_suffix=url_suffix,
params={"$select": "sharepointIds"},
)
return response.get("sharepointIds") or {}
def resolve_list_item_ids(self, site_id: str, item_id: str) -> tuple[str, str]:
"""Return the (list_id, list_item_unique_id) that address a driveItem as a list item.
Activities and analytics hang off the SharePoint list representation of an item, and
key off listItemUniqueId rather than the driveItem ID. Resolving here keeps the list
layer out of the command signatures.
Raises:
DemistoException: If the item exposes no SharePoint identifiers.
"""
sharepoint_ids = self.get_sharepoint_ids(site_id, item_id)
list_id = sharepoint_ids.get("listId")
list_item_unique_id = sharepoint_ids.get("listItemUniqueId")
if not list_id or not list_item_unique_id:
raise DemistoException(
f"Could not resolve the SharePoint list identifiers for item '{item_id}' in site '{site_id}'. "
f"This endpoint is only available for items stored in a SharePoint document library."
)
return list_id, list_item_unique_id
def list_driveitem_activities(self, site_id: str, item_id: str, next_page_url: str = "") -> dict:
"""List the activities that took place on a driveItem.
GET /sites/{site-id}/lists/{list-id}/items/{list-item-id}/activities, with the list
identifiers resolved from sharepointIds first.
This endpoint supports no OData query parameters, so there is no $top - result
limiting has to happen client-side. For the same reason its @odata.nextLink carries
an opaque token rather than $skiptoken, so url_validation is not applied here: it
would reject the very paging link the previous call returned.
"""
if next_page_url:
return self.ms_client.http_request(method="GET", full_url=next_page_url)
list_id, list_item_unique_id = self.resolve_list_item_ids(site_id, item_id)
url_suffix = f"sites/{site_id}/lists/{list_id}/items/{list_item_unique_id}/activities"
return self.ms_client.http_request(method="GET", url_suffix=url_suffix)
def get_driveitem_analytics(self, site_id: str, item_id: str, time_range: str) -> dict:
"""Retrieve activity statistics for a driveItem ('allTime' or 'lastSevenDays').
GET /sites/{site-id}/lists/{list-id}/items/{list-item-id}/analytics/{time_range},
same list-item addressing as activities. Note /sites/{id}/analytics is the analytics
of the site itself - a different resource.
"""
list_id, list_item_unique_id = self.resolve_list_item_ids(site_id, item_id)
url_suffix = f"sites/{site_id}/lists/{list_id}/items/{list_item_unique_id}/analytics/{time_range}"
return self.ms_client.http_request(method="GET", url_suffix=url_suffix)
def get_driveitem(
self,
object_type: str = "",
object_type_id: str = "",
item_id: str = "",
item_path: str = "",
share_url: str = "",
include_sharepoint_ids: bool = False,
) -> dict:
"""Retrieve the metadata of a single driveItem, with sharepointIds merged in.
Supports the three addressing modes Graph documents here: by item ID, by path under
the drive root, and by sharing URL. Exactly one is expected - the caller validates it.
sharepointIds is not returned even when asked for in $select, so it is fetched with a
second call. It carries the identifiers activities and analytics need.
"""
params = {"$select": DRIVEITEM_SELECT_FIELDS}
if share_url:
# The /shares route needs Files.ReadWrite.All (application) - a higher permission
# than the other two modes, even though this is a read.
url_suffix = f"shares/{encode_sharing_url(share_url)}/driveItem"
elif item_path:
url_suffix = self._driveitem_path_uri(object_type, object_type_id, item_path)
else:
url_suffix = self._driveitem_uri(object_type, object_type_id, item_id)
response = self.ms_client.http_request(method="GET", url_suffix=url_suffix, params=params)
if include_sharepoint_ids:
# For the share_url and path modes the caller may not know where the item lives,
# so take both identifiers from the response. driveId is what makes the lookup
# work for items outside the site's default document library.
parent_reference = response.get("parentReference") or {}
site_id = parent_reference.get("siteId")
drive_id = parent_reference.get("driveId")
resolved_item_id = response.get("id")
if resolved_item_id and (drive_id or site_id):
response["sharepointIds"] = self.get_sharepoint_ids(site_id or "", resolved_item_id, drive_id or "")
return response
def get_sensitivity_label(self, object_type: str, object_type_id: str, item_id: str) -> dict:
"""Retrieve the sensitivity label currently assigned to a drive item.
Args:
object_type: MS Graph resource. One of 'drives', 'groups', 'sites', 'users'.
object_type_id: MS Graph resource ID.
item_id: The drive item ID.
Returns:
Graph API raw response (the driveItem with the sensitivityLabel projection).
"""
uri = self._build_drive_item_uri(object_type, object_type_id, item_id)
return self.ms_client.http_request(
method="GET",
url_suffix=uri,
params={"$select": "sensitivityLabel"},
)
def assign_sensitivity_label(
self,
object_type: str,
object_type_id: str,
item_id: str,
sensitivity_label_id: str,
assignment_method: str,
justification_text: str,
) -> requests.Response:
"""Assign a sensitivity label to a drive item.
Args:
object_type: MS Graph resource. One of 'drives', 'groups', 'sites', 'users'.
object_type_id: MS Graph resource ID.
item_id: The drive item ID.
sensitivity_label_id: The GUID of the sensitivity label to assign. An empty
string instructs Microsoft Graph to remove the existing label.
assignment_method: One of 'standard', 'privileged', 'auto'.
justification_text: Free-text justification recorded with the assignment.
Returns:
Raw HTTP response object so the caller can inspect the status code and headers.
"""
uri = self._build_drive_item_uri(object_type, object_type_id, item_id) + "/assignSensitivityLabel"
body: dict = {"sensitivityLabelId": sensitivity_label_id}
if assignment_method:
body["assignmentMethod"] = assignment_method
if justification_text:
body["justificationText"] = justification_text
return self.ms_client.http_request(
method="POST",
url_suffix=uri,
json_data=body,
resp_type="response",
)
def test_function(client: MsGraphClient) -> str:
"""
Performs basic get request to get item samples
"""
response = "ok" if demisto.command() == "test-module" else "```✅ Success!```"
if (
demisto.params().get("self_deployed", False)
and demisto.command() == "test-module"
and (
client.ms_client.grant_type == AUTHORIZATION_CODE
or demisto.params().get("redirect_uri")
or demisto.params().get("auth_code_creds", {}).get("password", "")
)
):
raise DemistoException(
"The *Test* button is not available for the `self-deployed - Authorization Code Flow`.\n "
"Use the !msgraph-files-auth-test command instead "
"once all relevant parameters have been entered."
)
client.ms_client.http_request(url_suffix="sites", timeout=7, method="GET")
return response
def download_file_command(client: MsGraphClient, args: dict[str, str]) -> dict:
"""
This function runs download file command
:return: FileResult object
"""
object_type = args["object_type"]
validate_object_type(object_type)
object_type_id = args["object_type_id"]
item_id = args["item_id"]
file_name = args.get("file_name") or item_id
result = client.download_file(object_type=object_type, object_type_id=object_type_id, item_id=item_id)
return fileResult(file_name, result.content)
def list_drive_content_human_readable_object(parsed_drive_items: dict) -> dict:
return {
"Name": parsed_drive_items.get("Name"),
"ID": parsed_drive_items.get("ID"),
"CreatedBy": parsed_drive_items.get("CreatedBy", {}).get("DisplayName"),
"CreatedDateTime": parsed_drive_items.get("CreatedDateTime"),
"Description": parsed_drive_items.get("Description"),
"Size": parsed_drive_items.get("Size"),
"LastModifiedDateTime": parsed_drive_items.get("LastModifiedDateTime"),
"WebUrl": parsed_drive_items.get("WebUrl"),
}
def list_drive_content_command(client: MsGraphClient, args: dict[str, str]) -> tuple[str, dict, dict]:
"""
This function runs list drive children command
Note:
If the response does not contain results,
the Microsoft Graph API returns a dict with the key `@odata.null` set to true, indicating no data was found.
:return: human_readable, context, result
"""
object_type = args["object_type"]
validate_object_type(object_type)
object_type_id = args["object_type_id"]
item_id = args.get("item_id", "root")
limit = args.get("limit")
next_page_url = args.get("next_page_url")
result = client.list_drive_content(
object_type=object_type,
object_type_id=object_type_id,
item_id=item_id,
limit=limit,
next_page_url=next_page_url,
)
title = f"{INTEGRATION_NAME} - drivesItems information:"
parsed_drive_items = [parse_key_to_context(item) for item in result.get("value", [{}])]
human_readable_content = [list_drive_content_human_readable_object(item) for item in parsed_drive_items]
human_readable = tableToMarkdown(title, human_readable_content, headerTransform=pascalToSpace)
drive_items_outputs = {
"OdataContext": result["@odata.context"],
"Value": parsed_drive_items,
}
context = {
f"{INTEGRATION_NAME}.ListChildren(val.ItemID == obj.ItemID)": {
"ParentID": item_id,
"Children": drive_items_outputs,
"NextToken": result.get("@odata.nextLink"),
}
}
return human_readable, context, result
def list_share_point_sites_human_readable_object(parsed_drive_items: dict) -> dict:
return {
"Name": parsed_drive_items.get("Name"),
"ID": parsed_drive_items.get("ID"),
"CreatedDateTime": parsed_drive_items.get("CreatedDateTime"),
"LastModifiedDateTime": parsed_drive_items.get("LastModifiedDateTime"),
"WebUrl": parsed_drive_items.get("WebUrl"),
}
def list_sharepoint_sites_command(client: MsGraphClient, args: dict[str, str]) -> tuple[str, dict, dict]:
"""
This function runs list tenant site command
:return: human_readable, context, result
"""
keyword = args.get("keyword", "*")
result = client.list_sharepoint_sites(keyword)
parsed_sites_items = [parse_key_to_context(item) for item in result["value"]]
human_readable_content = [list_share_point_sites_human_readable_object(item) for item in parsed_sites_items]
context_entry = {
"OdataContext": result.get("@odata.context"),
"Value": parsed_sites_items,
}
context = {f"{INTEGRATION_NAME}.ListSites(val.ID === obj.ID)": context_entry}
title = "List Sites:"
human_readable = tableToMarkdown(title, human_readable_content, headerTransform=pascalToSpace)
return human_readable, context, result
def list_drives_human_readable_object(parsed_drive_items: dict) -> dict:
return {
"Name": parsed_drive_items.get("Name"),
"ID": parsed_drive_items.get("ID"),
"CreatedBy": parsed_drive_items.get("CreatedBy", {}).get("DisplayName"),
"CreatedDateTime": parsed_drive_items.get("CreatedDateTime"),
"Description": parsed_drive_items.get("Description"),
"DriveType": parsed_drive_items.get("DriveType"),
"LastModifiedDateTime": parsed_drive_items.get("LastModifiedDateTime"),
"WebUrl": parsed_drive_items.get("WebUrl"),
}
def list_drives_in_site_command(client: MsGraphClient, args: dict[str, str]) -> tuple[str, dict, dict]:
"""
This function run the list drives in site command
:return: human_readable, context, result
"""
site_id = args.get("site_id")
limit = args.get("limit")
next_page_url = args.get("next_page_url")
if next_page_url:
url_validation(next_page_url)
result = client.list_drives_in_site(site_id=site_id, limit=limit, next_page_url=next_page_url)
parsed_drive_items = [parse_key_to_context(item) for item in result["value"]]
human_readable_content = [list_drives_human_readable_object(item) for item in parsed_drive_items]
context_entry = {
"OdataContext": result.get("@odata.context"),
"Value": parsed_drive_items,
"NextToken": result.get("@odata.nextLink"),
}
title = f"{INTEGRATION_NAME} - Drives information:"
# Creating human readable for War room
human_readable = tableToMarkdown(title, human_readable_content, headerTransform=pascalToSpace)
# context == output
context = {f"{INTEGRATION_NAME}.ListDrives(val.ID === obj.ID)": context_entry}
return human_readable, context, result
def replace_an_existing_file_command(client: MsGraphClient, args: dict[str, str]) -> tuple[str, dict, dict]:
"""
This function runs the replace existing file command
:return: human_readable, context, result
"""
object_type = args["object_type"]
validate_object_type(object_type)
item_id = args["item_id"]
entry_id = args["entry_id"]
object_type_id = args["object_type_id"]
file_data, file_size, file_name = read_file(entry_id)
if file_size < client.MAX_ATTACHMENT_SIZE:
result = client.replace_existing_file(object_type, object_type_id, item_id, entry_id)
else:
result = client.replace_existing_file_with_upload_session(
object_type, object_type_id, item_id, entry_id, file_data, file_size, file_name
)
result = result.json()
demisto.debug(f"Response replace large existing file: \n {result} \n")
context_entry = parse_key_to_context(result)
human_readable_content = {
"ID": context_entry.get("ID"),
"Name": context_entry.get("Name"),
"CreatedBy": context_entry.get("CreatedBy", {}).get("DisplayName"),
"CreatedDateTime": context_entry.get("CreatedDateTime", {}),
"LastModifiedBy": context_entry.get("LastModifiedBy", {}).get("DisplayName"),
"Size": context_entry.get("Size"),
"WebUrl": context_entry.get("WebUrl"),
}
remove_nulls_from_dictionary(human_readable_content)
title = f"{INTEGRATION_NAME} - File information:"
# Creating human readable for War room
human_readable = tableToMarkdown(title, human_readable_content, headerTransform=pascalToSpace)
# context == output
context = {f"{INTEGRATION_NAME}.ReplacedFiles(val.ID === obj.ID)": context_entry}
return human_readable, context, result
def read_file(attach_id: str) -> tuple[bytes, int, str]:
"""
Reads file that was uploaded to War Room.
Args:
attach_id (str): The id of uploaded file to War Room.
Returns:
file_data (bytes): The file data.
file_size (int): The size of the file in bytes.
file_name (str): Uploaded file name.
"""
try:
file_info = demisto.getFilePath(attach_id)
with open(file_info["path"], "rb") as file_data:
file_data_read = file_data.read()
file_size = os.path.getsize(file_info["path"])
return file_data_read, file_size, file_info["name"]
except Exception as e:
raise Exception(f"Unable to read and decode in base 64 file with id {attach_id}", e) from e
def upload_new_file_command(client: MsGraphClient, args: dict[str, str]) -> tuple[str, dict, dict]:
"""
This function uploads new file to graph api
:return: human_readable, context, result
"""
object_type = args["object_type"]
validate_object_type(object_type)
object_type_id = args["object_type_id"]
parent_id = args["parent_id"]
entry_id = args["entry_id"]
file_data, file_size, _ = read_file(entry_id)
file_name = args["file_name"]
if file_size < client.MAX_ATTACHMENT_SIZE:
result = client.upload_new_file(object_type, object_type_id, parent_id, file_name, entry_id)
else:
result = client.upload_file_with_upload_session_flow(
object_type, object_type_id, parent_id, file_name, file_data, file_size
)
result = result.json()
demisto.debug(f"Response large file upload: \n {result} \n")
context_entry = parse_key_to_context(result)
human_readable_content = {
"ID": context_entry.get("ID"),
"Name": context_entry.get("Name"),
"CreatedBy": context_entry.get("CreatedBy", {}).get("DisplayName"),
"CreatedDateTime": context_entry.get("CreatedDateTime"),
"LastModifiedBy": context_entry.get("LastModifiedBy", {}).get("DisplayName"),
"Size": context_entry.get("Size"),
"WebUrl": context_entry.get("WebUrl"),
}
remove_nulls_from_dictionary(human_readable_content)
title = f"{INTEGRATION_NAME} - File information:"
# Creating human readable for War room
human_readable = tableToMarkdown(title, human_readable_content)
# context == output
context = {f"{INTEGRATION_NAME}.UploadedFiles(val.ID === obj.ID)": context_entry}
return human_readable, context, result
def create_new_folder_command(client: MsGraphClient, args: dict[str, str]) -> tuple[str, dict, dict]:
"""
This function runs create new folder command
:return: human_readable, context, result
"""
object_type = args["object_type"]
validate_object_type(object_type)
parent_id = args["parent_id"]
folder_name = args["folder_name"]
object_type_id = args["object_type_id"]
result = client.create_new_folder(object_type, object_type_id, parent_id, folder_name)
context_entry = parse_key_to_context(result)
human_readable_content = {
"ID": context_entry.get("ID"),
"Name": context_entry.get("Name"),
"CreatedBy": context_entry.get("CreatedBy", {}).get("DisplayName"),
"CreatedDateTime": context_entry.get("CreatedDateTime"),
"ChildCount": context_entry.get("Folder"),
"LastModifiedBy": context_entry.get("LastModifiedBy", {}).get("DisplayName"),
"Size": context_entry.get("Size"),
"WebUrl": context_entry.get("WebUrl"),
}
title = f"{INTEGRATION_NAME} - Folder information:"
# Creating human readable for War room
human_readable = tableToMarkdown(title, human_readable_content, headerTransform=pascalToSpace)
# context == output
context = {f"{INTEGRATION_NAME}.CreatedFolders(val.ID === obj.ID)": context_entry}
return human_readable, context, result
def delete_file_command(client: MsGraphClient, args: dict[str, str]) -> tuple[str, str]:
"""
runs delete file command
:return: raw response and action result test
"""
object_type = args["object_type"]
validate_object_type(object_type)
item_id = args["item_id"]
object_type_id = args["object_type_id"]
text = client.delete_file(object_type, object_type_id, item_id)
context_entry = text
title = f"{INTEGRATION_NAME} - File information:"
# Creating human readable for War room
human_readable = tableToMarkdown(title, context_entry, headers=item_id)
return human_readable, text # == raw response
def get_site_id_from_site_name(client: MsGraphClient, site_name: str | None) -> str:
"""
Retrieves the SharePoint site ID based on either the provided site ID or site name.
Args:
client (MsGraphClient): An instance of the Microsoft Graph client.
site_name (Optional[str]): The name of the SharePoint site.
Returns:
str: The SharePoint site ID.
Raises:
DemistoException: If neither site_id nor site_name is provided.
DemistoException: If the user has no permission to the site, the API returns a 404 error.
DemistoException: If the provided site name is invalid.
"""
if site_name:
try:
site_details = client.list_sharepoint_sites(site_name)["value"]
except DemistoException as e:
if e.res is not None and e.res.status_code == 404 and "Item not found" in e.res.text:
raise DemistoException(
f"Error getting site ID for {site_name}."
f" Ensure integration instance has permission for this site and site name is valid. Error details: {e}"
) from e
raise
if site_details:
return site_details[0]["id"]
else:
raise DemistoException(f"Site '{site_name}' not found. Please provide a valid site name.")
raise DemistoException("Please provide 'site_id' or 'site_name' parameter.")
def _md_parse_permission(permission: dict) -> dict:
"""Parses a permission dictionary into an output format.
Args:
permission: (dict) The permission dictionary to parse.
Returns:
dict: The parsed permission dictionary.
Example:
```
>>> permission = {
"id": "123",
"roles": ["role1", "role2"],
"grantedToIdentitiesV2": [{
"application": {
"displayName": "app1",
"id": "456"
}
}]
}
>>> permission_md = _md_parse_permission(permission)
>>> permission_md
{
"ID": "123",
"Roles": ["role1", "role2"],
"Application Name": ["app1"],
"Application ID": ["456"]
}
```
"""
identities: list[dict[str, dict[str, str]]] = permission.get("grantedToIdentitiesV2", [{}])
return {
"ID": permission.get("id"),
"Roles": permission.get("roles"),
"Application Name": [identity.get("application", {}).get("displayName") for identity in identities],
"Application ID": [identity.get("application", {}).get("id") for identity in identities],
}
def list_site_permissions_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""Lists permissions for a SharePoint site.
Args:
client (MsGraphClient): The Microsoft Graph client.
args (dict): The command arguments.
permission_id (Optional): The unique identifier for the permission.
limit (Optional): Number of items to return. Default is 50.
all_results (Optional): Return all results. Default is False.
site_name (Optional): The display name of the site. Either site_id or site_name is required.
site_id (Optional): The unique identifier for the site. Either site_id or site_name is required.
Returns:
CommandResults: The command results with outputs and readable output.
"""
permission_id = args.get("permission_id")
limit = arg_to_number(args.get("limit", 50))
all_results = argToBoolean(args.get("all_results", False))
site_id = args.get("site_id") or get_site_id_from_site_name(client, args.get("site_name"))
results = client.list_site_permissions(site_id, permission_id)
if permission_id:
return CommandResults(
outputs_prefix="MsGraphFiles.SitePermission",
outputs_key_field="id",
outputs=results,
readable_output=tableToMarkdown(name="Site Permission", t=_md_parse_permission(results)),
)
list_permissions = results.get("value", [])
return CommandResults(
outputs_prefix="MsGraphFiles.SitePermission",
outputs_key_field="id",
outputs=list_permissions if all_results else list_permissions[:limit],
readable_output=tableToMarkdown(
name="Site Permission", t=[_md_parse_permission(permission) for permission in list_permissions], removeNull=True
),
)
def create_site_permissions_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""Creates a new permission for a SharePoint site.
Args:
client (MsGraphClient): The Microsoft Graph client.
args (dict): The command arguments.
app_id (Required): The app ID to assign permissions to.
role (Required): The role(s) to assign. Can be "read", "write", or "owner".
display_name (Required): The display name for the permission.
site_name (Optional): The display name of the site. Either site_id or site_name is required.
site_id (Optional): The unique identifier for the site. Either site_id or site_name is required.
Returns:
CommandResults: The results including the new permission that was created.
"""
app_id = args["app_id"]
role = argToList(args["role"])
display_name = args["display_name"]
site_id = args.get("site_id") or get_site_id_from_site_name(client, args.get("site_name"))
results = client.create_site_permission(site_id, app_id, display_name, role)
return CommandResults(
outputs_prefix="MsGraphFiles.SitePermission",
outputs_key_field="id",
outputs=results,
readable_output=tableToMarkdown("Site Permission", t=_md_parse_permission(results)),
)
def update_site_permissions_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""Updates the permissions for a SharePoint site.
Args:
client (MsGraphClient): The Microsoft Graph client.
args (dict): The command arguments.
permission_id (Required): The ID of the permission to update.
role (Required): The updated role(s) for the permission. read/write/owner.
site_name (Optional): The display name of the site. Either site_id or site_name is required.
site_id (Optional): The unique identifier for the site. Either site_id or site_name is required.
Returns:
CommandResults: The command results with readable output.
"""
permission_id = args["permission_id"]
role = argToList(args["role"])
site_id = args.get("site_id") or get_site_id_from_site_name(client, args.get("site_name"))
results = client.update_site_permission(site_id, permission_id, role)
return CommandResults(
readable_output=(
f"Permission {permission_id} of site {site_id} was updated successfully with new role {results['roles']}."
)
)
def delete_site_permission_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""Deletes a permission from a SharePoint site.
Args:
client (MsGraphClient): The Microsoft Graph client.
args (dict): The command arguments.
permission_id (Required): The ID of the permission to delete.
site_name (Optional): The display name of the site. Either site_id or site_name is required.
site_id (Optional): The unique identifier for the site. Either site_id or site_name is required.
Returns:
CommandResults: The command results with readable output.
"""
permission_id = args["permission_id"]
site_id = args.get("site_id") or get_site_id_from_site_name(client, args.get("site_name"))
client.delete_site_permission(site_id, permission_id)
return CommandResults(readable_output="Site permission was deleted.")
def update_driveitem_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""Apply a PATCH on a driveItem (move within or across drives, rename, or update metadata).
Maps to Microsoft Graph: PATCH /v1.0/{drive-prefix}/items/{item-id}.
All body fields are optional per the Graph contract; only the keys whose argument
the caller provided are sent. Sending at least one update field is required.
Args:
client: The Microsoft Graph client.
args: Command arguments. See the YAML for the full list.
Returns:
CommandResults with the updated driveItem under MsGraphFiles.UpdatedItem.
"""
object_type = args["object_type"]
validate_object_type(object_type)
object_type_id = args["object_type_id"]
item_id = args["item_id"]
new_parent_id = args.get("new_parent_id", "")
new_parent_drive_id = args.get("new_parent_drive_id", "")
new_name = args.get("new_name", "")
description = args.get("description", "")
conflict_behavior = args.get("conflict_behavior", "")
body: dict = {}
parent_reference: dict = {}
if new_parent_id:
parent_reference["id"] = new_parent_id
if new_parent_drive_id:
parent_reference["driveId"] = new_parent_drive_id
if parent_reference:
body["parentReference"] = parent_reference
if new_name:
body["name"] = new_name
if description:
body["description"] = description
if conflict_behavior:
body["@microsoft.graph.conflictBehavior"] = conflict_behavior
if not body:
raise DemistoException(
"Provide at least one update field (new_parent_id, new_parent_drive_id, new_name, description, conflict_behavior)."
)
raw_response = client.update_driveitem(object_type, object_type_id, item_id, body)
context_entry = parse_key_to_context(raw_response)
human_readable_content = {
"ID": context_entry.get("ID"),
"Name": context_entry.get("Name"),
"LastModifiedDateTime": context_entry.get("LastModifiedDateTime"),
"ParentReferenceID": context_entry.get("ParentReference", {}).get("ID"),
"ParentReferenceDriveId": context_entry.get("ParentReference", {}).get("DriveId"),
"Size": context_entry.get("Size"),
"WebUrl": context_entry.get("WebUrl"),
}
remove_nulls_from_dictionary(human_readable_content)
readable_output = tableToMarkdown(
"Updated driveItem",
human_readable_content,
headerTransform=pascalToSpace,
removeNull=True,
)
return CommandResults(
outputs_prefix="MsGraphFiles.UpdatedItem",
outputs_key_field="ID",
outputs=context_entry,
raw_response=raw_response,
readable_output=readable_output,
)
def copy_driveitem_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""Initiate an asynchronous driveItem copy and return the monitor URL.
Maps to Microsoft Graph: POST /v1.0/{drive-prefix}/items/{item-id}/copy.
Microsoft Graph performs the copy asynchronously and responds 202 Accepted with a Location
header pointing to a monitor URL. The caller is expected to poll the monitor URL against
Microsoft Graph until the copy reaches a terminal status.
All body fields and the conflict_behavior query parameter are optional per the Graph contract;
only keys whose argument the caller provided are sent.
Args:
client: The Microsoft Graph client.
args: Command arguments. See the YAML for the full list.
Returns:
CommandResults with MsGraphFiles.CopyOperation context including MonitorUrl.
"""
object_type = args["object_type"]
validate_object_type(object_type)
object_type_id = args["object_type_id"]
item_id = args["item_id"]
destination_parent_id = args.get("destination_parent_id", "")
destination_drive_id = args.get("destination_drive_id", "")
new_name = args.get("new_name", "")
conflict_behavior = args.get("conflict_behavior", "")
children_only_raw = args.get("children_only", "")
body: dict = {}
parent_reference: dict = {}
if destination_parent_id:
parent_reference["id"] = destination_parent_id
if destination_drive_id:
parent_reference["driveId"] = destination_drive_id
if parent_reference:
body["parentReference"] = parent_reference
if new_name:
body["name"] = new_name
if children_only_raw:
body["childrenOnly"] = argToBoolean(children_only_raw)
params: dict = {}
if conflict_behavior:
params["@microsoft.graph.conflictBehavior"] = conflict_behavior
response = client.copy_driveitem(object_type, object_type_id, item_id, body, params)
monitor_url = response.headers.get("Location", "")
outputs = {
"MonitorUrl": monitor_url,
"ItemId": item_id,
"ObjectType": object_type,
"ObjectTypeId": object_type_id,
}
readable_output = tableToMarkdown(
"Copy operation accepted",
{"MonitorUrl": monitor_url},
removeNull=True,
)
readable_output += (
"\nPoll the MonitorUrl directly against Microsoft Graph "
"(using the same access token) until status is `completed` or `failed`."
)
return CommandResults(
outputs_prefix="MsGraphFiles.CopyOperation",
outputs_key_field="ItemId",
outputs=outputs,
readable_output=readable_output,
)
def _decode_sharepoint_login_name(login_name: str) -> str:
"""Extract a human-readable identifier from a SharePoint claims-encoded loginName.
Microsoft Graph encodes external/guest users using the SharePoint claims format
"i:0#.f|membership|<encoded-upn>". For guest users the encoded UPN looks like
"<email-with-underscore>#ext#@<tenant>.onmicrosoft.com", where the original "@" in the
guest's email is replaced with "_". This helper returns the decoded email when the input
matches that pattern, otherwise returns the input unchanged so callers can still surface
something meaningful.
Args:
login_name: The raw SharePoint claims-encoded loginName string.
Returns:
The decoded email (best effort) or the original loginName.
"""
if not login_name or "|" not in login_name:
return login_name
encoded_upn = login_name.rsplit("|", 1)[-1]
if "#ext#" in encoded_upn:
# Guest user: "<email-with-underscore>#ext#@<tenant>.onmicrosoft.com"
guest_part = encoded_upn.split("#ext#", 1)[0]
# The original "@" was replaced with the LAST "_" in the email.
if "_" in guest_part:
local, _, domain = guest_part.rpartition("_")
return f"{local}@{domain}"
return guest_part
return encoded_upn
def _lookup(source: dict, *names: str) -> Any:
"""Return the first truthy value among 'names', matching keys case-insensitively.
Graph payloads reach these helpers in two casings: parse_key_to_context title-cases
('Email'), but does not recurse into lists, so identities nested in a list such as
grantedToIdentitiesV2 keep Microsoft's camelCase ('email'). Matching case-insensitively
handles both without every call site spelling out each key twice.
"""
if not isinstance(source, dict):
return None
lowered = {key.lower(): value for key, value in source.items()}
for name in names:
value = lowered.get(name.lower())
if value:
return value
return None
def _identity_label(identity: dict) -> str:
"""Return the best human-readable label for an IdentitySet identity (user/siteUser/group/application/device).
Prefers an email, then a loginName (decoded if SharePoint claims-encoded), then a
displayName, then an ID. Returns an empty string when nothing useful is present.
"""
if email := _lookup(identity, "email"):
return email
if login_name := _lookup(identity, "loginName"):
return _decode_sharepoint_login_name(login_name)
return _lookup(identity, "displayName", "id") or ""
def _summarize_identity_set(identity_set: dict) -> str:
"""Return a label for an identitySet, for example the 'actor' of an itemActivity.
An identitySet nests the identity under a role key, so this unwraps the first populated
one and labels it. Unlike _identity_label, which expects an already-unwrapped identity.
"""
for role in IDENTITY_ROLE_KEYS:
if label := _identity_label(_lookup(identity_set, role) or {}):
return label
return ""
def _summarize_activity_actions(activity: dict) -> str:
"""Summarize which action facets an itemActivity represents, comma-separated.
v1.0 nests the facets under 'action', e.g. {"edit": {}, "version": {"newVersion": "2.0"}}.
The facet *names* carry the information - most have an empty object as their value. Any
facet Graph adds later is surfaced automatically.
"""
action = activity.get("Action")
if not isinstance(action, dict):
return ""
labels = []
for name in sorted(action):
details = action[name]
# 'version' is the one facet with a useful payload, so include the resulting version.
new_version = details.get("NewVersion") if isinstance(details, dict) else None
labels.append(f"{name} ({new_version})" if new_version else name)
return ", ".join(labels)
def _activity_recorded_time(activity: dict) -> str:
"""Return the ISO timestamp of an itemActivity, or an empty string.
v1.0 carries it as 'times.recordedDateTime'. 'activityDateTime' is accepted as a fallback
because it appears on the beta endpoint and in parts of the documentation.
"""
times = activity.get("Times")
if isinstance(times, dict) and times.get("RecordedDateTime"):
return times["RecordedDateTime"]
return activity.get("ActivityDateTime") or ""
def _summarize_permission_grantees(perm: dict) -> str:
"""Build a comma-separated label of all grantees on a single permission entry.
Walks every identity source Microsoft Graph may populate on a driveItem permission:
grantedToV2 / grantedTo (single IdentitySet), grantedToIdentitiesV2 / grantedToIdentities
(list of IdentitySets). Inside each IdentitySet it considers user, siteUser, group,
application and device. Duplicates are removed while preserving order.
"""
identity_sets: list[dict] = []
single = _lookup(perm, "grantedToV2", "grantedTo")
if isinstance(single, dict):
identity_sets.append(single)
listed = _lookup(perm, "grantedToIdentitiesV2", "grantedToIdentities")
if isinstance(listed, list):
identity_sets.extend(item for item in listed if isinstance(item, dict))
labels: list[str] = []
seen: set[str] = set()
for identity_set in identity_sets:
for role in IDENTITY_ROLE_KEYS:
label = _identity_label(_lookup(identity_set, role) or {})
if label and label not in seen:
labels.append(label)
seen.add(label)
return ", ".join(labels)
def list_driveitem_permissions_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""List permissions (sharing entries) on a driveItem.
Maps to Microsoft Graph: GET /v1.0/{drive-prefix}/items/{item-id}/permissions.
Args:
client: The Microsoft Graph client.
args: Command arguments. See the YAML for the full list.
Returns:
CommandResults with MsGraphFiles.ItemPermission context.
"""
object_type = args["object_type"]
validate_object_type(object_type)
object_type_id = args["object_type_id"]
item_id = args["item_id"]
limit = args.get("limit") or None
next_page_url = args.get("next_page_url") or None
raw_response = client.list_driveitem_permissions(
object_type=object_type,
object_type_id=object_type_id,
item_id=item_id,
limit=limit,
next_page_url=next_page_url,
)
parsed_permissions = [parse_key_to_context(p) for p in raw_response.get("value", [])]
outputs = {
"Value": parsed_permissions,
"ItemId": item_id,
"ObjectType": object_type,
"ObjectTypeId": object_type_id,
"OdataContext": raw_response.get("@odata.context"),
"NextToken": raw_response.get("@odata.nextLink"),
}
remove_nulls_from_dictionary(outputs)
readable_rows = [
{
"ID": perm.get("ID"),
"Roles": perm.get("Roles"),
"LinkScope": (perm.get("Link") or {}).get("Scope"),
"LinkType": (perm.get("Link") or {}).get("Type"),
"GrantedTo": _summarize_permission_grantees(perm),
"InheritedFrom": "yes" if perm.get("InheritedFrom") else None,
}
for perm in parsed_permissions
]
readable_output = tableToMarkdown(
"DriveItem permissions",
readable_rows,
headerTransform=pascalToSpace,
removeNull=True,
)
return CommandResults(
outputs_prefix="MsGraphFiles.ItemPermission",
outputs_key_field="ItemId",
outputs=outputs,
raw_response=raw_response,
readable_output=readable_output,
)
def delete_driveitem_permission_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""Delete (revoke) a single sharing permission on a driveItem.
Maps to Microsoft Graph: DELETE /v1.0/{drive-prefix}/items/{item-id}/permissions/{perm-id}.
Microsoft Graph returns 204 No Content. Errors (including 404 itemNotFound) are surfaced
verbatim. For bulk-delete loops where intermittent 404s are acceptable, set "Continue on
error" on the calling task.
Args:
client: The Microsoft Graph client.
args: Command arguments. See the YAML for the full list.
Returns:
CommandResults with MsGraphFiles.RemovedItemPermission echo context.
"""
object_type = args["object_type"]
validate_object_type(object_type)
object_type_id = args["object_type_id"]
item_id = args["item_id"]
permission_id = args["permission_id"]
client.delete_driveitem_permission(object_type, object_type_id, item_id, permission_id)
outputs = {
"ItemId": item_id,
"PermissionId": permission_id,
"ObjectType": object_type,
"ObjectTypeId": object_type_id,
}
readable_output = tableToMarkdown(
"Permission removed",
{"ItemId": item_id, "PermissionId": permission_id},
)
return CommandResults(
outputs_prefix="MsGraphFiles.RemovedItemPermission",
outputs_key_field="PermissionId",
outputs=outputs,
readable_output=readable_output,
)
def _nested_identity_set(raw_identity_set: Any) -> dict:
"""Convert a raw Graph identitySet into context conventions, keeping the role nesting.
parse_key_to_context() routes CreatedBy and LastModifiedBy through remove_identity_key(),
which collapses {'user': {...}} into {'DisplayName': ..., 'Type': 'User'} and discards the
email and ID. Those two fields are the join keys between this command and the activities
command - an actor is correlated by email, not by display name - so CRTX-218926 specifies
the full nested identity here. The flattening stays in place for the older commands that
already depend on it; only this command re-derives the nested shape from the raw response.
Returns:
{'User': {'Email': ..., 'ID': ..., 'DisplayName': ...}}, with one key per role present.
Returns {} when the payload carries no usable identity.
"""
if not isinstance(raw_identity_set, dict):
return {}
return {
string_to_context_key(camel_case_to_underscore(role)): parse_key_to_context(identity)
for role, identity in raw_identity_set.items()
if isinstance(identity, dict) and identity
}
def _driveitem_metadata_readable(context_entry: dict) -> str:
"""Render the human-readable table for a single driveItem's metadata.
ListItemUniqueId is surfaced alongside the plain metadata because it is the identifier
the activities and analytics commands need, and is otherwise buried in sharepointIds.
DriveId is shown because the ID above it is only meaningful within that drive.
"""
sharepoint_ids = context_entry.get("SharepointIds") or {}
human_readable_content = {
"ID": context_entry.get("ID"),
"Name": context_entry.get("Name"),
"Size": context_entry.get("Size"),
"DriveId": context_entry.get("DriveId"),
"CreatedDateTime": context_entry.get("CreatedDateTime"),
"LastModifiedDateTime": context_entry.get("LastModifiedDateTime"),
"CreatedBy": (context_entry.get("CreatedBy") or {}).get("User", {}).get("DisplayName"),
"LastModifiedBy": (context_entry.get("LastModifiedBy") or {}).get("User", {}).get("DisplayName"),
"ListItemUniqueId": sharepoint_ids.get("ListItemUniqueId"),
"WebUrl": context_entry.get("WebUrl"),
}
remove_nulls_from_dictionary(human_readable_content)
return tableToMarkdown(
"DriveItem metadata",
human_readable_content,
headerTransform=pascalToSpace,
removeNull=True,
)
def get_driveitem_metadata_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""Retrieve the metadata of a single driveItem (file or folder).
Maps to GET /v1.0/{drive-prefix}/items/{item-id}, /root:/{item-path} or
/shares/{encoded-sharing-url}/driveItem, depending on the addressing argument supplied.
Exactly one is required; YAML cannot express that rule, so it is enforced here.
The returned ItemID is scoped to the drive that hosts the item, which is why DriveId is
surfaced next to it: a sharing URL to a personal file resolves against that user's
OneDrive, so its ItemID differs from the one the same file has under a site library.
"""
addressing = resolve_item_addressing(args, allow_share_url=True)
include_sharepoint_ids = argToBoolean(args.get("include_sharepoint_ids", "false"))
raw_response = client.get_driveitem(
object_type=addressing["object_type"],
object_type_id=addressing["object_type_id"],
item_id=addressing["value"] if addressing["mode"] == "item_id" else "",
item_path=addressing["value"] if addressing["mode"] == "item_path" else "",
share_url=addressing["value"] if addressing["mode"] == "share_url" else "",
include_sharepoint_ids=include_sharepoint_ids,
)
context_entry = parse_key_to_context(raw_response)
# Restore the nested identity that parse_key_to_context() flattened away. Without this the
# creator's email and ID are lost, and the readable table below finds nothing under 'User'.
for raw_field, context_field in (("createdBy", "CreatedBy"), ("lastModifiedBy", "LastModifiedBy")):
if identity_set := _nested_identity_set(raw_response.get(raw_field)):
context_entry[context_field] = identity_set
# SiteID and DriveId are surfaced at the top level because callers need them for follow-up
# commands, and they are otherwise buried inside parentReference. DriveId also states which
# drive ItemID belongs to - the two are only meaningful together.
parent_reference = context_entry.get("ParentReference") or {}
context_entry["SiteID"] = parent_reference.get("SiteId")
context_entry["DriveId"] = parent_reference.get("DriveId")
context_entry["ItemID"] = context_entry.get("ID")
remove_nulls_from_dictionary(context_entry)
return CommandResults(
outputs_prefix="MsGraphFiles.Files",
outputs_key_field="ID",
outputs=context_entry,
raw_response=raw_response,
readable_output=_driveitem_metadata_readable(context_entry),
)
def _driveitem_activities_readable(parsed_activities: list, item_id: str) -> str:
"""Render the activities table, or a plain message when there are none."""
if not parsed_activities:
return f"No activities were found for item {item_id}."
readable_rows = [
{
"ID": activity.get("ID"),
"RecordedDateTime": _activity_recorded_time(activity),
"Actor": _summarize_identity_set(activity.get("Actor") or {}),
"Action": _summarize_activity_actions(activity),
}
for activity in parsed_activities
]
return tableToMarkdown(
f"Activities for item {item_id}",
readable_rows,
headers=["ID", "RecordedDateTime", "Actor", "Action"],
headerTransform=pascalToSpace,
removeNull=True,
)
def list_driveitem_activities_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""List the activities that took place on a driveItem.
Maps to GET /v1.0/sites/{site-id}/lists/{list-id}/items/{list-item-id}/activities.
Activities hang off the SharePoint list representation, but resolving the list
identifiers is internal - callers supply only site_id and item_id.
The endpoint accepts no OData query parameters, so 'limit' is applied client-side
rather than sent as $top.
"""
site_id = args["site_id"]
item_id = args["item_id"]
next_page_url = args.get("next_page_url") or ""
limit = arg_to_number(args.get("limit"))
if limit is not None and limit <= 0:
raise DemistoException(f"The limit argument must be a positive integer. Got {limit}.")
raw_response = client.list_driveitem_activities(
site_id=site_id,
item_id=item_id,
next_page_url=next_page_url,
)
activities = raw_response.get("value", [])
if limit:
activities = activities[:limit]
parsed_activities = [parse_key_to_context(activity) for activity in activities]
outputs = {
"Value": parsed_activities,
"ItemId": item_id,
"SiteID": site_id,
"OdataContext": raw_response.get("@odata.context"),
"NextToken": raw_response.get("@odata.nextLink"),
}
remove_nulls_from_dictionary(outputs)
return CommandResults(
outputs_prefix="MsGraphFiles.ItemActivity",
outputs_key_field="ItemId",
outputs=outputs,
raw_response=raw_response,
readable_output=_driveitem_activities_readable(parsed_activities, item_id),
)
def _driveitem_analytics_readable(parsed_stats: dict, item_id: str, time_range: str) -> str:
"""Render the analytics table, or a plain message when no data was returned.
Every action facet present in the response is surfaced, so a facet Microsoft adds later
appears without a code change - the same approach _summarize_activity_actions takes.
An absent facet simply produces no row. The itemActivityStat scalars are skipped: they
describe the window the statistics cover, not an action.
"""
non_action_fields = {"StartDateTime", "EndDateTime", "IsTrending", "IncompleteData"}
readable_rows = [
{
"Action": action,
"ActionCount": details.get("ActionCount"),
"ActorCount": details.get("ActorCount"),
}
for action, details in sorted(parsed_stats.items())
if action not in non_action_fields and isinstance(details, dict)
]
if not readable_rows:
return (
f"No analytics data was returned for item {item_id} over '{time_range}'. "
f"This can also mean the tenant plan does not surface analytics data."
)
return tableToMarkdown(
f"Analytics for item {item_id} ({time_range})",
readable_rows,
headerTransform=pascalToSpace,
removeNull=True,
)
def get_driveitem_analytics_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""Retrieve activity statistics (views, edits) for a driveItem.
Maps to GET /v1.0/sites/{site-id}/lists/{list-id}/items/{list-item-id}/analytics/
{time_range}, same internal list-item addressing as the activities command.
/sites/{id}/analytics is site-level analytics - a different resource - and is not used.
"""
site_id = args["site_id"]
item_id = args["item_id"]
time_range = args.get("time_range") or "allTime"
if time_range not in ("allTime", "lastSevenDays"):
raise DemistoException(f"Invalid time_range: {time_range}. Allowed values are: allTime, lastSevenDays.")
raw_response = client.get_driveitem_analytics(
site_id=site_id,
item_id=item_id,
time_range=time_range,
)
# For allTime the stats are nested under the time-range key; other shapes are returned bare.
# Normalize both into a single stats object.
stats = raw_response.get(time_range) or raw_response
parsed_stats = parse_key_to_context(stats)
outputs = {
"ItemId": item_id,
"SiteID": site_id,
"TimeRange": time_range,
"Stats": parsed_stats,
}
remove_nulls_from_dictionary(outputs)
return CommandResults(
outputs_prefix="MsGraphFiles.ItemAnalytics",
outputs_key_field="ItemId",
outputs=outputs,
raw_response=raw_response,
readable_output=_driveitem_analytics_readable(parsed_stats, item_id, time_range),
)
def get_sensitivity_label_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""Retrieve the sensitivity label currently assigned to a drive item.
A missing or null `sensitivityLabel` field in the Graph response is the documented
"no label assigned" case and is returned as a successful result with empty label
fields, not an error.
Args:
client: The Microsoft Graph client.
args: The command arguments.
object_type (Required): The MS Graph resource. One of drives, groups, sites, users.
object_type_id (Required): The MS Graph resource ID.
item_id (Required): The drive item ID.
Returns:
CommandResults with the drive item ID and all sensitivity label fields
as returned by Microsoft Graph.
"""
object_type = args.get("object_type", "")
validate_object_type(object_type)
object_type_id = args.get("object_type_id", "")
item_id = args.get("item_id", "")
raw_response = client.get_sensitivity_label(object_type, object_type_id, item_id)
label = raw_response.get("sensitivityLabel") or {}
outputs: dict = {"itemId": item_id}
outputs.update(label)
if label:
readable_output = tableToMarkdown(
"Sensitivity Label",
outputs,
headerTransform=pascalToSpace,
)
else:
readable_output = f"No sensitivity label is assigned to drive item `{item_id}`."
return CommandResults(
outputs_prefix="MsGraphFiles.SensitivityLabel",
outputs_key_field="itemId",
outputs=outputs,
readable_output=readable_output,
raw_response=raw_response,
)
def assign_sensitivity_label_command(client: MsGraphClient, args: dict[str, str]) -> CommandResults:
"""Assign a sensitivity label to a drive item.
Microsoft Graph treats `assignSensitivityLabel` as a long-running operation and
returns `202 Accepted` with a `Location` response header pointing to the operation
status URL. The handler surfaces that URL verbatim in `outputs["location"]`. Non-2xx
responses raise an exception that is caught by the outer `main()` try/except and
surfaced via `return_error` with the raw Graph error message.
Args:
client: The Microsoft Graph client.
args: The command arguments.
object_type (Required): The MS Graph resource. One of drives, groups, sites, users.
object_type_id (Required): The MS Graph resource ID.
item_id (Required): The drive item ID.
sensitivity_label_id (Optional): The GUID of the sensitivity label to assign.
An empty string instructs Microsoft Graph to remove the existing label.
assignment_method (Optional): One of standard, privileged, auto.
justification_text (Optional): Free-text justification recorded with the
assignment.
Returns:
CommandResults with the drive item ID, the assigned label GUID, and the
`Location` header URL returned by Microsoft Graph.
"""
object_type = args.get("object_type", "")
validate_object_type(object_type)
object_type_id = args.get("object_type_id", "")
item_id = args.get("item_id", "")
sensitivity_label_id = args.get("sensitivity_label_id", "")
assignment_method = args.get("assignment_method", "")
justification_text = args.get("justification_text", "")
response = client.assign_sensitivity_label(
object_type=object_type,
object_type_id=object_type_id,
item_id=item_id,
sensitivity_label_id=sensitivity_label_id,
assignment_method=assignment_method,
justification_text=justification_text,
)
response_headers = getattr(response, "headers", {}) or {}
location = response_headers.get("Location") or ""
outputs = {
"itemId": item_id,
"sensitivityLabelId": sensitivity_label_id,
"location": location,
}
readable_output = tableToMarkdown(
name="Assigned Sensitivity Label",
t=outputs,
headers=["itemId", "sensitivityLabelId", "location"],
headerTransform=pascalToSpace,
)
return CommandResults(
outputs_prefix="MsGraphFiles.AssignedSensitivityLabel",
outputs_key_field="itemId",
outputs=outputs,
readable_output=readable_output,
)
def run_microsoft_graph_files_integration():
params: dict = demisto.params()
args = demisto.args()
command = demisto.command()
base_url: str = params.get("host", "https://graph.microsoft.com").rstrip("/") + "/v1.0/"
tenant = params.get("credentials_tenant_id", {}).get("password") or params.get("tenant_id")
auth_id = params.get("credentials_auth_id", {}).get("password") or params.get("auth_id")
enc_key = params.get("credentials_enc_key", {}).get("password") or params.get("enc_key")
use_ssl: bool = not params.get("insecure", False)
proxy: bool = params.get("proxy", False)
ok_codes: tuple = (200, 201, 202, 204)
certificate_thumbprint = params.get("credentials_certificate_thumbprint", {}).get("password") or params.get(
"certificate_thumbprint"
)
private_key = params.get("private_key")
managed_identities_client_id: Optional[str] = get_azure_managed_identities_client_id(params)
self_deployed: bool = params.get("self_deployed", False) or managed_identities_client_id is not None
auth_code = params.get("auth_code_creds", {}).get("password", "")
redirect_uri = params.get("redirect_uri", "")
try:
client = MsGraphClient(
base_url=base_url,
tenant_id=tenant,
auth_id=auth_id,
enc_key=enc_key,
app_name=APP_NAME,
verify=use_ssl,
proxy=proxy,
self_deployed=self_deployed,
ok_codes=ok_codes,
certificate_thumbprint=certificate_thumbprint,
private_key=private_key,
managed_identities_client_id=managed_identities_client_id,
redirect_uri=redirect_uri,
auth_code=auth_code,
)
demisto.debug(f"Command being called is {command}")
if command == "test-module":
# This is the call made when pressing the integration Test button.
return_results(test_function(client))
elif command == "msgraph-files-auth-test":
return_results(test_function(client))
elif command == "msgraph-files-generate-login-url":
return_results(generate_login_url(client.ms_client))
elif command == "msgraph-files-auth-reset":
return_results(reset_auth())
elif command == "msgraph-delete-file":
readable_output, raw_response = delete_file_command(client, args)
return_outputs(readable_output=readable_output, raw_response=raw_response)
elif command == "msgraph-list-sharepoint-sites":
return_outputs(*list_sharepoint_sites_command(client, args))
elif command == "msgraph-download-file":
# it has to be demisto.results instead of return_outputs.
# because fileResult contains 'content': '' and if that key is empty return_outputs returns error.
demisto.results(download_file_command(client, args))
elif command == "msgraph-list-drive-content":
return_outputs(*list_drive_content_command(client, args))
elif command == "msgraph-create-new-folder":
return_outputs(*create_new_folder_command(client, args))
elif command == "msgraph-replace-existing-file":
return_outputs(*replace_an_existing_file_command(client, args))
elif command == "msgraph-list-drives-in-site":
return_outputs(*list_drives_in_site_command(client, args))
elif command == "msgraph-upload-new-file":
return_outputs(*upload_new_file_command(client, args))
elif command == "msgraph-list-site-permissions":
return_results(list_site_permissions_command(client, args))
elif command == "msgraph-create-site-permissions":
return_results(create_site_permissions_command(client, args))
elif command == "msgraph-update-site-permissions":
return_results(update_site_permissions_command(client, args))
elif command == "msgraph-delete-site-permissions":
return_results(delete_site_permission_command(client, args))
elif command == "msgraph-driveitem-update":
return_results(update_driveitem_command(client, args))
elif command == "msgraph-driveitem-copy":
return_results(copy_driveitem_command(client, args))
elif command == "msgraph-driveitem-permissions-list":
return_results(list_driveitem_permissions_command(client, args))
elif command == "msgraph-driveitem-permission-delete":
return_results(delete_driveitem_permission_command(client, args))
elif command == "msgraph-driveitem-metadata-get":
return_results(get_driveitem_metadata_command(client, args))
elif command == "msgraph-driveitem-activities-list":
return_results(list_driveitem_activities_command(client, args))
elif command == "msgraph-driveitem-analytics-get":
return_results(get_driveitem_analytics_command(client, args))
elif command == "msgraph-get-sensitivity-label":
return_results(get_sensitivity_label_command(client, args))
elif command == "msgraph-assign-sensitivity-label":
return_results(assign_sensitivity_label_command(client, args))
else:
raise NotImplementedError(f"Command {command} is not implemented")
except Exception as e:
return_error(f"Failed to execute {command} command.\nError:\n{e!s}")
README
To use the common Microsoft Graph Files integration logic, run the following command to import the MicrosoftGraphFilesApiModule.
def main():
run_microsoft_graph_files_integration()
from MicrosoftGraphFilesApiModule import * # noqa: E402
if __name__ in ["builtins", "__main__"]:
main()
Then, the run_microsoft_graph_files_integration entry point and MsGraphClient class will be available for usage. For the canonical consumer, see the Microsoft Graph Files integration.