<jupyter_start><jupyter_text>Automation Gallery - Credential Scan on Azure Blob Storage__Notebook Version:__ 1.0__Python Version:__ Python 3.8 - AzureML__Required Packages:__ No__Platforms Supported:__  Azure Machine Learning Notebooks     __Data Source Required:__ No      DescriptionThis notebook provides step-by-step instructions and sample code to detect credential leak into Azure Blob Storage using Azure SDK for Python.*** No need to download and install any other Python modules. ****** Please run the cells sequentially to avoid errors.  Please do not use "run all cells". ***  Table of Contents1. Warm-up2. Authentication to Azure Storage3. Scan Azure Blob for Leaking Credentials  1. Warm-up<jupyter_code># If you need to know what Python modules are available, you may run this:
#help("modules")

!pip install azure-storage-blob
# Load Python libraries that will be used in this notebook
from azure.mgmt.storage import StorageManagementClient
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient, __version__
from azure.mgmt.resource import ResourceManagementClient

from azure.identity import ManagedIdentityCredential
from datetime import datetime, timedelta, timezone

import json
import os
import csv
from IPython.display import display, HTML, Markdown
import re
# Functions will be used in this notebook
def read_config_values(file_path):
    "This loads pre-generated parameters for Sentinel Workspace"
    with open(file_path) as json_file:
        if json_file:
            json_config = json.load(json_file)
            return (json_config["tenant_id"],
                    json_config["subscription_id"],
                    json_config["resource_group"],
                    json_config["workspace_id"],
                    json_config["workspace_name"],
                    json_config["user_alias"],
                    json_config["user_object_id"])
    return None



def get_file_content(blob):
    "Decoding file content"
    try:
        content = blob.content_as_text(max_concurrency=1, encoding='UTF-8')
    except UnicodeDecodeError:
        content = blob.content_as_text(max_concurrency=1, encoding='UTF-16')
    return content

def get_regex_list():
    "This function return RegEx list for credscan"
    regex_list = [
        "(?i)(ida:password|IssuerSecret|(api|client|app(lication)?)[_\\- ]?(key|secret)[^,a-z]|\\.azuredatabricks\\.net).{0,10}(dapi)?[a-z0-9/+]{22}",
        "(?i)(x-api-(key|token).{0,10}[a-z0-9/+]{40}|v1\\.[a-z0-9/+]{40}[^a-z0-9/+])",
        "(?-i:)\\WAIza(?i)[a-z0-9_\\\\\\-]{35}\\W",
        "(?i)(\\Wsig\\W|Secret(Value)?|IssuerSecret|(\\Wsas|primary|secondary|management|Shared(Access(Policy)?)?).?Key|\\.azure\\-devices\\.net|\\.(core|servicebus|redis\\.cache|accesscontrol|mediaservices)\\.(windows\\.net|chinacloudapi\\.cn|cloudapi\\.de|usgovcloudapi\\.net)|New\\-AzureRedisCache).{0,100}([a-z0-9/+]{43}=)",
        "(?i)visualstudio\\.com.{1,100}\\W(?-i:)[a-z2-7]{52}\\W",
        "(?i)se=2021.+sig=[a-z0-9%]{43,63}%3d",
        "(?i)(x-functions-key|ApiKey|Code=|\\.azurewebsites\\.net/api/).{0,100}[a-z0-9/\\+]{54}={2}",
        "(?i)code=[a-z0-9%]{54,74}(%3d){2}",
        "(?i)(userpwd|publishingpassword).{0,100}[a-z0-9/\\+]{60}\\W",
        "(?i)[^a-z0-9/\\+][a-z0-9/\\+]{86}==",
        "(?-i:)\\-{5}BEGIN( ([DR]SA|EC|OPENSSH|PGP))? PRIVATE KEY( BLOCK)?\\-{5}",
        "(?i)(app(lication)?|client)[_\\- ]?(key(url)?|secret)([\\s=:>]{1,10}|[\\s\"':=|>\\]]{3,15}|[\"'=:\\(]{2})[^\\-]",
        "(?i)refresh[_\\-]?token([\\s=:>]{1,10}|[\\s\"':=|>\\]]{3,15}|[\"'=:\\(]{2})(\"data:text/plain,.+\"|[a-z0-9/+=_.-]{20,200})",
        "(?i)AccessToken(Secret)?([\\s\"':=|>\\]]{3,15}|[\"'=:\\(]{2}|[\\s=:>]{1,10})[a-z0-9/+=_.-]{20,200}",
        "(?i)[a-z0-9]{3,5}://[^%:\\s\"'/][^:\\s\"'/\\$]+[^:\\s\"'/\\$%]:([^%\\s\"'/][^@\\s\"'/]{0,100}[^%\\s\"'/])@[\\$a-z0-9:\\.\\-_%\\?=/]+",
        "(?i)snmp(\\-server)?\\.exe.{0,100}(priv|community)",
        "(?i)(ConvertTo\\-?SecureString\\s*((\\(|\\Wstring)\\s*)?['\"]+)",
        "(?i)(Consumer|api)[_\\- ]?(Secret|Key)([\\s=:>]{1,10}|[\\s\"':=|>,\\]]{3,15}|[\"'=:\\(]{2})[^\\s]{5,}",
        "(?i)authorization[,\\[:= \"']+([dbaohmnsv])",
        "(?i)-u\\s+.{2,100}-p\\s+[^\\-/]",
        "(?i)(amqp|ssh|(ht|f)tps?)://[^%:\\s\"'/][^:\\s\"'/\\$]+[^:\\s\"'/\\$%]:([^%\\s\"'/][^@\\s\"'/]{0,100}[^%\\s\"'/])@[\\$a-z0-9:\\.\\-_%\\?=/]+",
        "(?i)(\\Waws|amazon)?.{0,5}(secret|access.?key).{0,10}\\W[a-z0-9/\\+]{40}",
        "(?-i:)(eyJ0eXAiOiJKV1Qi|eyJhbGci)",
        "(?i)@(\\.(on)?)?microsoft\\.com[ -~\\s]{1,100}?(\\w?pass\\w?)",
        "(?i)net(\\.exe)?.{1,5}(user\\s+|share\\s+/user:|user-?secrets? set)\\s+[a-z0-9]",
        "(?i)xox[pbar]\\-[a-z0-9]",
        "(?i)[\":\\s=]((x?corp|extranet(test)?|ntdev)(\\.microsoft\\.com)?|corp|redmond|europe|middleeast|northamerica|southpacific|southamerica|fareast|africa|exchange|extranet(test)?|partners|parttest|ntdev|ntwksta)\\W.{0,100}(password|\\<PASSWORD>|\\<PASSWORD>|\\<PASSWORD>|userpass)",
        "(?i)(sign_in|SharePointOnlineAuthenticatedContext|(User|Exchange)Credentials?|password)[ -~\\s]{0,100}?@([a-z0-9.]+\\.(on)?)?microsoft\\.com['\"]?",
        "(?i)(\\.database\\.azure\\.com|\\.database(\\.secure)?\\.windows\\.net|\\.cloudapp\\.net|\\.database\\.usgovcloudapi\\.net|\\.database\\.chinacloudapi\\.cn|\\.database.cloudapi.de).{0,100}(DB_PASS|(sql|service)?password|\\Wpwd\\W)",
        "(?i)(secret(.?key)?|password)[\"']?\\s*[:=]\\s*[\"'][^\\s]+?[\"']",
        "(?i)[^a-z\\$](DB_USER|user id|uid|(sql)?user(name)?|service\\s?account)\\s*[^\\w\\s,]([ -~\\s]{2,120}?|[ -~]{2,30}?)([^a-z\\s\\$]|\\s)\\s*(DB_PASS|(sql|service)?password|pwd)",
        "(?i)(password|secret(key)?)[ \\t]*[=:]+[ \\t]*([^:\\s\"';,<]{2,200})",
    ]

    return regex_list

def set_continuation_flag(flag):
    if flag == False:
        print("continuation flag is false.")
    return flag

def convert_result_to_string(result_row):
    if (type(result_row)) == str:
        return result_row
    elif (type(result_row)) == tuple:
        return ','.join([m for m in result_row if len(m) > 0])

def export_csv(file_name, data_list):
    with open(file_name, 'w') as f:
        w = csv.writer(f, delimiter = ',')
        w.writerows([x.split(',') for x in data_list])

def file_modified_date_check(days_back, modified_date):
    aware_local_now = datetime.now(timezone.utc).astimezone()
    time_between_modified = aware_local_now - modified_date
    return time_between_modified.days < days_back
# Calling the above function to populate Sentinel workspace parameters
# The file, config.json, was generated by the system, however, you may modify the values, or manually set the variables
tenant_id, subscription_id, resource_group, workspace_id, workspace_name, user_alias, user_object_id = read_config_values('config.json');<jupyter_output><empty_output><jupyter_text>2. Authentication to Azure Storage<jupyter_code>client_id = os.environ.get('DEFAULT_IDENTITY_CLIENT_ID')
credential = ManagedIdentityCredential(client_id=client_id)
token = credential.get_token('https://storage.azure.com/')
resource_client = ResourceManagementClient(credential, subscription_id)
storage_client = StorageManagementClient(credential, subscription_id)<jupyter_output><empty_output><jupyter_text>3. Scan Azure Blob for Leaking Credentials<jupyter_code>group_name = 'sentinel_notebooks'
storage_account_name = 'credscantest'
container_name = 'samples'
storage_keys = storage_client.storage_accounts.list_keys(group_name,storage_account_name)
if storage_keys != None:
    storage_key = {v.key_name: v.value for v in storage_keys.keys}['key1']
    print(storage_key)

    blob_service_client = BlobServiceClient(
        account_url="https://{0}.blob.core.windows.net".format(storage_account_name),
        credential=storage_key
    )

    if blob_service_client != None:
        container_client = blob_service_client.get_container_client(container_name)
        if container_client != None:
            blob_list = container_client.list_blobs()
            for b in blob_list:
                print(b.last_modified)
                if (file_modified_date_check(2, b.last_modified)):
                    blob = container_client.download_blob(b)
                    content = get_file_content(blob)
                
                    # Run Regex strings on the file content
                    import warnings
                    warnings.filterwarnings('ignore')
                    result_list = []
                    csv_string = ""
                    if content != None:
                        has_leaking = False
                        regex_list = get_regex_list()
                        for regex in regex_list:
                            re.compile(regex)
                            results = re.findall(regex, content)
                            if results:
                                print("================================================")
                                print("MATCHED REGEX:\n" + regex)
                                print("------------------------------------------------")
                                print("FILE: " + b.name + "\n")
                                #print(content)
                                print("---------------MATCHED CONTENT -----------------")
                                for result in results:
                                    print(str(result))
                                    csv_string = convert_result_to_string(result)
                                    result_list.append(csv_string)   
                                print("================================================")
                                has_leaking = True   
                        if has_leaking == False:
                            print('No leaking data found')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.22207687663513787, 'fraction_numerical': 0.02183537935198229, 'mean_word_length': 2.651359294636297, 'pattern_counts': {'":': 1, '<': 18, '<?xml version=': 0, '>': 25, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '550049', 'license': 'MIT', 'n_tokens_mistral': 3573, 'n_tokens_neox': 3334, 'n_words': 484, 'path': 'Automation Gallery - Credential Scan on Azure Blob Storage.ipynb', 'repo_name': 'argvssarma/AzureSentinel', 'size': '9923'}