olipassey/AzAppRegistrationExpiry
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

export

Browse Source
This commit is contained in:
Oli Passey
2025-01-17 16:19:29 +00:00
parent 7572413871
commit 1bf3df60ad
8 changed files with 370 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
data_export.py
View File

@@ -187,6 +187,18 @@ def generate_expiry_text(app_name, days_to_expiry, expiry_date):
</div>
"""
def sort_app_registrations(app_registrations):
current_date = datetime.utcnow()
for app in app_registrations:
expiry_date_str = app["passwordCredentials"][0]["endDateTime"]
expiry_date = datetime.strptime(expiry_date_str.split('.')[0], '%Y-%m-%dT%H:%M:%S')
days_to_expiry = (expiry_date - current_date).days
app["days_to_expiry"] = days_to_expiry
app["expiry_date"] = expiry_date
sorted_apps = sorted(app_registrations, key=lambda x: x["days_to_expiry"], reverse=False)
return sorted_apps
# Example usage
if __name__ == "__main__":
# Sample app registration data
@@ -197,10 +209,21 @@ if __name__ == "__main__":
},
{
"displayName": "App2",
"passwordCredentials": [{"endDateTime": "2023-12-31T23:59:59.9999999Z"}]
},
{
"displayName": "App3",
"passwordCredentials": [{"endDateTime": "2025-01-15T23:59:59.9999999Z"}]
}
]
sorted_apps = sort_app_registrations(app_registrations)
html_output = ""
for app in sorted_apps:
html_output += generate_expiry_text(app["displayName"], app["days_to_expiry"], app["expiry_date"])
print(html_output)
# Write to JSON
write_to_json(app_registrations)

75
getdrive.py Normal file
View File

@@ -0,0 +1,75 @@
import os
import requests
import logging
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(asctime)s - %(message)s')
def get_access_token():
client_id = os.getenv('AZURE_CLIENT_ID')
client_secret = os.getenv('AZURE_CLIENT_SECRET')
tenant_id = os.getenv('AZURE_TENANT_ID')
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
token_data = {
'grant_type': 'client_credentials',
'client_id': client_id,
'client_secret': client_secret,
'scope': 'https://graph.microsoft.com/.default'
}
token_response = requests.post(token_url, data=token_data)
access_token = token_response.json().get('access_token')
if not access_token:
logging.error("Failed to obtain access token")
logging.error(token_response.json())
return None
return access_token
def list_drive_items(site_id, drive_id, parent_id=None):
access_token = get_access_token()
if not access_token:
return None
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
if parent_id:
items_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}/items/{parent_id}/children"
else:
items_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}/root/children"
response = requests.get(items_url, headers=headers)
if response.status_code != 200:
logging.error(f"Failed to list drive items: {response.status_code}")
logging.error(response.json())
return None
items = response.json().get('value', [])
if not items:
logging.error("No items found in the drive")
return None
for item in items:
logging.info(f"Found item: {item.get('name')} with ID: {item.get('id')}")
if item.get('folder'):
# Recursively list items in the folder
list_drive_items(site_id, drive_id, item.get('id'))
return items
if __name__ == "__main__":
site_id = "opassey.sharepoint.com,8e038e2a-139b-4e46-893a-bcf76062e063,5dcd71be-16b7-42ba-add6-4068b88ef3aa"
drive_id = os.getenv('SHAREPOINT_DRIVE_ID')
items = list_drive_items(site_id, drive_id)
if items:
logging.info("Listed drive items successfully")
else:
logging.error("Failed to list drive items")

66
getsites.py Normal file
View File

@@ -0,0 +1,66 @@
import os
import requests
import logging
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def get_access_token():
client_id = os.getenv('AZURE_CLIENT_ID')
client_secret = os.getenv('AZURE_CLIENT_SECRET')
tenant_id = os.getenv('AZURE_TENANT_ID')
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
token_data = {
'grant_type': 'client_credentials',
'client_id': client_id,
'client_secret': client_secret,
'scope': 'https://graph.microsoft.com/.default'
}
token_response = requests.post(token_url, data=token_data)
access_token = token_response.json().get('access_token')
if not access_token:
logging.error("Failed to obtain access token")
logging.error(token_response.json())
return None
return access_token
def list_sites():
access_token = get_access_token()
if not access_token:
return None
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
sites_url = "https://graph.microsoft.com/v1.0/sites?search=*"
response = requests.get(sites_url, headers=headers)
if response.status_code != 200:
logging.error(f"Failed to list sites: {response.status_code}")
logging.error(response.json())
return None
sites = response.json().get('value', [])
if not sites:
logging.error("No sites found")
return None
for site in sites:
logging.info(f"Found site: {site.get('name')} with ID: {site.get('id')}")
return sites
if __name__ == "__main__":
sites = list_sites()
if sites:
logging.info("Listed sites successfully")
else:
logging.error("Failed to list sites")

67
gettoken.py Normal file
View File

@@ -0,0 +1,67 @@
import os
import requests
import logging
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def get_access_token():
client_id = os.getenv('AZURE_CLIENT_ID')
client_secret = os.getenv('AZURE_CLIENT_SECRET')
tenant_id = os.getenv('AZURE_TENANT_ID')
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
token_data = {
'grant_type': 'client_credentials',
'client_id': client_id,
'client_secret': client_secret,
'scope': 'https://graph.microsoft.com/.default'
}
token_response = requests.post(token_url, data=token_data)
access_token = token_response.json().get('access_token')
if not access_token:
logging.error("Failed to obtain access token")
logging.error(token_response.json())
return None
return access_token
def get_drive_id(site_id):
access_token = get_access_token()
if not access_token:
return None
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
drives_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives"
response = requests.get(drives_url, headers=headers)
if response.status_code != 200:
logging.error(f"Failed to get drives: {response.status_code}")
logging.error(response.json())
return None
drives = response.json().get('value', [])
if not drives:
logging.error("No drives found")
return None
# Assuming you want the first drive in the list
drive_id = drives[0].get('id')
logging.info(f"Found drive ID: {drive_id}")
return drive_id
if __name__ == "__main__":
site_id = os.getenv('SHAREPOINT_SITE_ID')
drive_id = get_drive_id(site_id)
if drive_id:
logging.info(f"Drive ID: {drive_id}")
else:
logging.error("Failed to obtain drive ID")

2
main.py
View File

@@ -1,5 +1,5 @@
from azure_client import get_app_registrations
from sharepoint_client import store_app_registrations # Uncomment this line
from sharepoint_client import store_app_registrations
from notification import send_notifications
from data_export import write_to_json, generate_html

2
notification.py
View File

@@ -30,7 +30,7 @@ def send_notifications(app_registrations):
# Get the current date
current_date = datetime.utcnow()
notification_periods = [60, 30, 7, 1]
notification_periods = [60, 30, 7, 3, 2, 1]
# Generate HTML content
html_content = generate_html(app_registrations)

89
sharepoint_client.py
View File

@@ -11,14 +11,62 @@ load_dotenv()
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def update_excel_file(file_id, data):
def get_access_token():
client_id = os.getenv('AZURE_CLIENT_ID')
client_secret = os.getenv('AZURE_CLIENT_SECRET')
tenant_id = os.getenv('AZURE_TENANT_ID')
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
token_data = {
'grant_type': 'client_credentials',
'client_id': client_id,
'client_secret': client_secret,
'scope': 'https://graph.microsoft.com/.default'
}
token_response = requests.post(token_url, data=token_data)
access_token = token_response.json().get('access_token')
if not access_token:
logging.error("Failed to obtain access token")
logging.error(token_response.json())
return None
return access_token
def get_drive_id(site_id):
access_token = get_access_token()
if not access_token:
return None
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
drives_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives"
response = requests.get(drives_url, headers=headers)
if response.status_code != 200:
logging.error(f"Failed to get drives: {response.status_code}")
logging.error(response.json())
return None
drives = response.json().get('value', [])
if not drives:
logging.error("No drives found")
return None
# Assuming you want the first drive in the list
drive_id = drives[0].get('id')
logging.info(f"Found drive ID: {drive_id}")
return drive_id
def update_excel_file(site_id, drive_id, file_id, data):
"""
Update Excel Online file with new data using OneDrive for Business API
"""
client_id = os.getenv('AZURE_CLIENT_ID')
client_secret = os.getenv('AZURE_CLIENT_SECRET')
tenant_id = os.getenv('AZURE_TENANT_ID')
user_email = os.getenv('USER_EMAIL')
# Get access token
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
@@ -30,7 +78,11 @@ def update_excel_file(file_id, data):
}
token_response = requests.post(token_url, data=token_data)
access_token = token_response.json()['access_token']
access_token = token_response.json().get('access_token')
if not access_token:
logging.error("Failed to obtain access token")
logging.error(token_response.json())
return
headers = {
'Authorization': f'Bearer {access_token}',
@@ -38,21 +90,14 @@ def update_excel_file(file_id, data):
}
try:
# Get the drive ID first
drives_url = f"https://graph.microsoft.com/v1.0/users/{user_email}/drive"
drive_response = requests.get(drives_url, headers=headers)
drive_response.raise_for_status()
drive_id = drive_response.json().get('id')
logging.info(f"Found drive ID: {drive_id}")
# Get the file details to verify it exists
file_url = f"https://graph.microsoft.com/v1.0/drives/{drive_id}/items/{file_id}"
file_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}/items/{file_id}"
file_response = requests.get(file_url, headers=headers)
file_response.raise_for_status()
logging.info(f"Found file: {file_response.json().get('name')}")
# Clear existing data (except header)
clear_url = f"https://graph.microsoft.com/v1.0/drives/{drive_id}/items/{file_id}/workbook/worksheets/Sheet1/range(address='A2:D1000')"
clear_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}/items/{file_id}/workbook/worksheets/Sheet1/range(address='A2:D1000')"
clear_data = {
"values": [[""]*4]*999
}
@@ -61,7 +106,7 @@ def update_excel_file(file_id, data):
logging.info("Cleared existing data from Excel file")
# Write new data
update_url = f"https://graph.microsoft.com/v1.0/drives/{drive_id}/items/{file_id}/workbook/worksheets/Sheet1/range(address='A2:D{len(data)+1}')"
update_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}/items/{file_id}/workbook/worksheets/Sheet1/range(address='A2:D{len(data)+1}')"
update_data = {
"values": data
}
@@ -89,7 +134,7 @@ def store_app_registrations(app_registrations):
expiry_date = password_credentials[0].get('endDateTime')
if expiry_date:
if (expiry_date.endswith('ZZ')):
if expiry_date.endswith('ZZ'):
expiry_date = expiry_date[:-1]
elif expiry_date.endswith('Z'):
expiry_date = expiry_date[:-1]
@@ -116,10 +161,20 @@ def store_app_registrations(app_registrations):
# Update Excel file
try:
excel_file_id = os.getenv('EXCEL_FILE_ID')
update_excel_file(excel_file_id, excel_data)
site_id = os.getenv('SHAREPOINT_SITE_ID')
drive_id = os.getenv('SHAREPOINT_DRIVE_ID')
file_id = os.getenv('EXCEL_FILE_ID')
update_excel_file(site_id, drive_id, file_id, excel_data)
logging.info("Successfully updated Excel file")
except Exception as e:
logging.error(f"Failed to update Excel file: {e}")
logging.info("Finished processing app registrations")
logging.info("Finished processing app registrations")
if __name__ == "__main__":
site_id = os.getenv('SHAREPOINT_SITE_ID')
drive_id = get_drive_id(site_id)
if drive_id:
logging.info(f"Drive ID: {drive_id}")
else:
logging.error("Failed to obtain drive ID")

64
sys_status.py Normal file
View File

@@ -0,0 +1,64 @@
import os
from datetime import datetime, timedelta
from email.mime.text import MIMEText
from email.utils import formataddr
import smtplib
import logging
def check_client_id_expiry():
# Load environment variables
client_id = os.getenv('AZURE_CLIENT_ID')
client_secret = os.getenv('AZURE_CLIENT_SECRET')
tenant_id = os.getenv('AZURE_TENANT_ID')
smtp_server = os.getenv('SMTP_SERVER')
smtp_port = int(os.getenv('SMTP_PORT'))
smtp_username = os.getenv('SMTP_USERNAME')
smtp_password = os.getenv('SMTP_PASSWORD')
from_email = os.getenv('FROM_EMAIL')
from_name = os.getenv('FROM_NAME')
admin_email = os.getenv('ADMIN_EMAIL')
# Check if the client ID is expiring soon
expiry_date_str = os.getenv('CLIENT_ID_EXPIRY_DATE')
if not expiry_date_str:
logging.error("CLIENT_ID_EXPIRY_DATE not set in environment variables")
return
try:
expiry_date = datetime.strptime(expiry_date_str, '%Y-%m-%d')
except ValueError as e:
logging.error(f"Error parsing CLIENT_ID_EXPIRY_DATE: {e}")
return
days_to_expiry = (expiry_date - datetime.utcnow()).days
if days_to_expiry <= 30:
subject = "Warning: Azure Client ID Expiry Notification"
body = f"""
<html>
<body>
<p>The Azure Client ID <strong>{client_id}</strong> is set to expire in
<span style="color: red; font-weight: bold;">{days_to_expiry} days</span>
on <strong>{expiry_date.strftime('%Y-%m-%d')}</strong>.</p>
<p>Please take the necessary actions to renew the client ID before it expires.</p>
</body>
</html>
"""
# Create email message
msg = MIMEText(body, 'html')
msg['Subject'] = subject
msg['From'] = formataddr((from_name, from_email))
msg['To'] = admin_email
try:
with smtplib.SMTP(smtp_server, smtp_port) as server:
server.starttls()
server.login(smtp_username, smtp_password)
server.sendmail(from_email, admin_email, msg.as_string())
logging.info(f"Successfully sent client ID expiry warning to {admin_email}")
except Exception as e:
logging.error(f"Failed to send client ID expiry warning: {e}")
if __name__ == "__main__":
check_client_id_expiry()