Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update User Audit Scripts for Pagination #240

Merged
merged 4 commits into from
Oct 3, 2024
Merged

Update User Audit Scripts for Pagination #240

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
70210f7
Fix the script
ababhis2 Oct 2, 2024
e265405
all scripts done
ababhis2 Oct 2, 2024
8a2cd03
Remove debug statement
ababhis2 Oct 2, 2024
82ae43b
Address comment
ababhis2 Oct 3, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 37 additions & 27 deletions User Audit/audit_test_full.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,15 +32,28 @@ def flatten(x, name='', exclude=exclude):

token = os.environ.get('API_KEY')
base_url = "https://api.kennasecurity.com"
users_url= base_url + "/users"
per_page = 10
users_url= base_url + "/users?per_page=" + str(per_page)
roles_url = base_url + "/roles"
audit_logs_url = base_url + "/audit_logs/"

headers = {"Accept": "application/json", "X-Risk-Token":token}
pages = -1
page = 1
users_df = pd.DataFrame()

while True:
paged_url = users_url + "&page=" + str(page)
print("Requesting data from", paged_url)
users_response = requests.get(paged_url, headers=headers).json()
users_df = pd.concat([users_df, pd.DataFrame(json_normalize([flatten_json(x) for x in users_response['users']]))], ignore_index=True)
# do this once
if 'meta' in users_response and pages == -1:
pages = users_response['meta']['pages']
if page >= pages:
break
page += 1

users_response = requests.get(users_url, headers=headers).json()

users_df = pd.DataFrame(json_normalize([flatten_json(x) for x in users_response['users']]))
users_df = users_df.rename(columns={"id":"user_id","created_at":"user_created_at","updated_at":"user_updated_at"})

users_df['user_created_at'] = pd.to_datetime(users_df['user_created_at'], format='%Y-%m-%d', errors='coerce').dt.date
Expand Down Expand Up @@ -126,36 +139,33 @@ def flatten(x, name='', exclude=exclude):
audit_logs_processed.append(event_data)

audit_df = pd.DataFrame(audit_logs_processed)

# Open the existing workbook with 'openpyxl'
wb = load_workbook('cvm_user_audit.xlsx')
users_sheet = wb['Users']

# Write the 'Audit Logs' DataFrame to the workbook
with pd.ExcelWriter('cvm_user_audit.xlsx', engine='openpyxl') as writer:
writer.book = wb
merged_df = pd.merge(users_df, audit_df, left_on="user_id", right_on="kenna_user_id", how="inner")
merged_df.to_excel(writer, sheet_name='Audit Logs')
# Define a fill for highlighting cells
green_fill = PatternFill(start_color='00FF00', end_color='00FF00', fill_type='solid')

# Get the 'Users' sheet
users_sheet = wb['Users']
# Get emails from audit logs data with 'source' as 'API'
api_emails = [log['user_email'] for log in audit_logs_processed if log['source'] == 'API']

# Define a fill for highlighting cells
green_fill = PatternFill(start_color='00FF00', end_color='00FF00', fill_type='solid')
# Iterate over the rows in the 'Users' DataFrame
for i, row in users_df.iterrows():
email = row['email'] # Assuming 'email' is a column in your DataFrame
# Check if the email is in api_emails
if email in api_emails:
# If it is, highlight the entire row
for j in range(1, len(row) + 1):
print("Highlighting")
users_sheet.cell(row=i+2, column=j).fill = green_fill # i+2 because DataFrame is 0-indexed and Worksheet is 1-indexed, and we have a header row

# Get emails from audit logs data with 'source' as 'API'
api_emails = [log['user_email'] for log in audit_logs_processed if log['source'] == 'API']
# Save the workbook
wb.save('cvm_user_audit.xlsx')

# Iterate over the rows in the 'Users' DataFrame
for i, row in users_df.iterrows():
email = row['email'] # Assuming 'email' is a column in your DataFrame
# Check if the email is in api_emails
if email in api_emails:
# If it is, highlight the entire row
for j in range(1, len(row) + 1):
users_sheet.cell(row=i+2, column=j).fill = green_fill # i+2 because DataFrame is 0-indexed and Worksheet is 1-indexed, and we have a header row
# Write the 'Audit Logs' DataFrame to the workbook
with pd.ExcelWriter('cvm_user_audit.xlsx', engine='openpyxl', mode='a') as writer:
merged_df = pd.merge(users_df, audit_df, left_on="user_id", right_on="kenna_user_id", how="inner")
merged_df.to_excel(writer, sheet_name='Audit Logs')

# Save the workbook
wb.save('cvm_user_audit.xlsx')

print('User, role, and audit log data has been saved to the file cvm_user_audit.xlsx.')
print('Users that have never logged in are highlighted in red. Users that have not logged in for over 30 days are highlighted in yellow.')
Expand Down
35 changes: 23 additions & 12 deletions User Audit/useraudit.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@ def flatten_json(nested_json, exclude=['roles']):
The flattened json object if successful, None otherwise.
"""
out = {}

def flatten(x, name='', exclude=exclude):
if type(x) is dict:
for a in x:
Expand All @@ -31,27 +30,42 @@ def flatten(x, name='', exclude=exclude):
return out

token = sys.argv[1]
base_url = "https://api.kennasecurity.com"
users_url= base_url + "/users"
# increase per_page to get more data per request
per_page = 10
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can this be configurable like user can adjust this number?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

these scripts are sent to the client so either CX or client can customize the scripts. I have added a small README snippet about per_page but I do not see the need to make it a CLI option. I do not like the way CLI is being used in user audit script but not planning to change it

base_url = "http://api.kennasecurity.com"
users_url= base_url + "/users?per_page=" + str(per_page)
roles_url = base_url + "/roles"

#print(users_url)

headers = {"Accept": "application/json", "X-Risk-Token":token, "User-Agent": 'user_audit/1.0.0 (Kenna Security)'}

users_response = requests.get(users_url, headers=headers).json()
headers = {"Accept": "*/*", "X-Risk-Token":token, "User-Agent": 'PostmanRuntime/7.42.0'}

#print(users_response)
writer = pd.ExcelWriter('kenna_user_audit.xlsx', engine='xlsxwriter', date_format='m/d/yyyy')
pages = -1
page = 1
users_df = pd.DataFrame()

while True:
paged_url = users_url + "&page=" + str(page)
print("Requesting data from", paged_url)
users_response = requests.get(paged_url, headers=headers).json()
users_df = pd.concat([users_df, pd.DataFrame(json_normalize([flatten_json(x) for x in users_response['users']]))], ignore_index=True)
# do this once
if 'meta' in users_response and pages == -1:
pages = users_response['meta']['pages']
if page >= pages:
break
page += 1

users_df = pd.DataFrame(json_normalize([flatten_json(x) for x in users_response['users']]))
users_df = users_df.rename(columns={"id":"user_id","created_at":"user_created_at","updated_at":"user_updated_at"})

users_df['user_created_at'] = pd.to_datetime(users_df['user_created_at'], format='%Y-%m-%d', errors='coerce').dt.date
users_df['last_sign_in_at'] = pd.to_datetime(users_df['last_sign_in_at'], format='%Y-%m-%d', errors='coerce').dt.date

print('Printing Users data sample:')
print(users_df.head(2))

users_df.to_excel(writer, sheet_name='Users')
#remove comments to troubleshoot columns
#for col in users_df.columns:
#print(col)
Expand All @@ -70,11 +84,8 @@ def flatten(x, name='', exclude=exclude):
#for col in roles_df.columns:
#print(col)

writer = pd.ExcelWriter('kenna_user_audit.xlsx', engine='xlsxwriter', date_format='m/d/yyyy')

workbook = writer.book

users_df.to_excel(writer, sheet_name='Users')
roles_df.to_excel(writer, sheet_name='Roles')

red_format = workbook.add_format({'bg_color': '#FFC7CE','font_color': '#9C0006'})
Expand All @@ -89,7 +100,7 @@ def flatten(x, name='', exclude=exclude):
users_sheet.conditional_format('$J$2:$J$99999', {'type': 'blanks', 'format': red_format})
users_sheet.conditional_format('$J$2:$J$99999', {'type': 'formula', 'criteria': '=J2<TODAY()-30', 'format': yellow_format})

writer.save()
writer.close()

print('User and role data has been saved to the file kenna_user_audit.xlsx.')
print('Users that have never logged in are highlighted in red. Users that have not logged in for over 30 days are highlighted in yellow.')
Loading