Hello @Ruben Cocheno,
I managed to create the below Python script to collect top identities from my Umbrella Org using the Legacy Umbrella API: Top Identities (all) - Cloud Security API - Cisco DevNet. However, I don't get even 2% of the total top identities. Can you help me with this weird behavior!?
import os
import time
from datetime import datetime, timedelta
import requests
from dotenv import load_dotenv
from oauthlib.oauth2 import BackendApplicationClient
from requests.auth import AuthBase
from requests.auth import HTTPBasicAuth as BasicAuth
from requests_oauthlib import OAuth2Session
from rich import print
from urllib3 import disable_warnings
from urllib3.exceptions import InsecureRequestWarning
load_dotenv()
disable_warnings(category=InsecureRequestWarning)
LEGACY_CLIENT_ID = os.getenv("LEGACY_CLIENT_ID", None)
LEGACY_CLIENT_SECRET = os.getenv("LEGACY_CLIENT_SECRET", None)
# BearerAuth
class BearerAuth(AuthBase):
def __init__(self, token: str):
self.token = token
def __call__(self, r):
r.headers["Authorization"] = f"Bearer {self.token}"
r.headers["Accept"] = "application/json"
r.headers["Content-Type"] = "application/json"
return r
start = time.perf_counter() # Start time
print("Generating Umbrella access token...", end="\r")
s = requests.Session()
# Authorization
try:
client = BackendApplicationClient(client_id=LEGACY_CLIENT_ID)
oauth = OAuth2Session(client=client)
# Access Token
token = oauth.fetch_token(
token_url="https://management.api.umbrella.com/auth/v2/oauth2/token",
client_id=LEGACY_CLIENT_ID,
client_secret=LEGACY_CLIENT_SECRET,
auth=BasicAuth(LEGACY_CLIENT_ID, LEGACY_CLIENT_SECRET),
verify=False,
)
except requests.HTTPError as e:
err_msg = (
f'[red]{e.response.status_code} - {e.response.json().get("data").get("error")}'
)
raise SystemExit(print(err_msg)) from e
else:
print("[green]Generated access token successfully")
access_token = token.get("access_token")
print(f"Fetching Umbrella Top Identities...", end="\r")
try:
# GET: top identites (all)
res = s.get(
url=f"https://reports.api.umbrella.com/v2/organizations/{os.getenv('UMBRELLA_ORG_ID')}/top-identities",
params={"identitytypes": "mobile_device,android"},
auth=BearerAuth(access_token),
allow_redirects=False,
)
res.raise_for_status()
except requests.HTTPError as e:
err_msg = f"[red]{e.response.status_code} - {e.response.json().get('data').get('error')}"
raise SystemExit(print(err_msg)) from e
else:
sample_set = set() # Create a sample_set for unique labels
total_requests = 0 # Total requests counter
offset = 30 # Assuming 30 pages
months = 27 # Duration of 27 months
for i in range(0, offset):
start_date = 1633039200000 # 1 October 2021 12:00 AM (Local time)
for _ in range(0, months):
end_date = start_date + 2592000000000 # End date is start + 30 days
# Another request to handle redirection
r = s.get(
url=res.headers.get("location"), # Get redirection URL
params={
"from": start_date,
"to": end_date,
"limit": 5000,
"offset": i,
},
auth=BearerAuth(access_token),
)
start_date += 2592000000
print(f"{r.status_code} - {r.reason}", r.request.url)
if r.ok:
total_requests += 1
for identity in r.json().get("data"):
# Check condition to add the identity to the set
if (
len(identity.get("identity").get("label")) == 15
and identity.get("identity").get("label").isnumeric()
):
# Add new unique indentity labels to the sample set
sample_set.add(identity.get("identity").get("label"))
else:
print(f"{r.status_code} - {r.reason} - {r.json()}")
# Create text output file
with open(
f"Unique-Top-Identities_All_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.txt",
"wt",
) as f:
for identity in sample_set:
# Write identity in a .txt file
f.write(f"{identity},\n")
print(
f"[green]Created {f.name} file with {len(sample_set):,} unique identities from a total of {total_requests:,} requests"
)
print(f"EET: {timedelta(seconds=time.perf_counter() - start)}")
Osama Abbas
