emitlinks/glpi-airwatch-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Updated both scripts to use AirwatchAPI and GLPIAPI modules

Browse Source
This commit is contained in:
Jason Secula
2025-07-05 13:09:28 +02:00
parent a4ee8a1ae8
commit 9f04afb153
8 changed files with 526 additions and 543 deletions
Download Patch File Download Diff File Expand all files Collapse all files

448
scripts/syncGLPI.py
View File

@ -1,78 +1,75 @@
#!/usr/bin/python3
import os
import base64
import requests
import json
import argparse
from cryptography.hazmat.primitives.serialization import pkcs12, pkcs7
from cryptography.hazmat.primitives import hashes, serialization
import logging
from datetime import datetime
from functions import getSettings
from includes.airwatchAPI import *
from includes.GLPIAPI import *
parser = argparse.ArgumentParser()
parser.add_argument("-debug", action=argparse.BooleanOptionalAction)
parser.add_argument("-searchFilter", type=str, choices=["Id", "SerialNumber", "Imei", "UserName"])
parser.add_argument("-searchValue", type=str)
parser.add_argument("-force", action=argparse.BooleanOptionalAction)
parser.add_argument("-sF", "--searchFilter", dest="searchfilter", type=str, choices=["Id", "SerialNumber", "Imei", "UserName"])
parser.add_argument("-sV","--searchValue", dest="searchvalue", type=str)
parser.add_argument("-c", "--configPath", dest="configpath", type=str)
parser.add_argument("-f", "--force", dest="force", action="store_true")
parser.add_argument("-s", "--silent", dest="silent", action="store_true")
parser.add_argument("-v", "--verbose", dest="debug", action="store_true")
args = parser.parse_args()
settingsDefault = {
"airwatchServer":"https://airwatchServer",
"airwatchAPIKey":"APIKEY",
"airwatchAuthMethod":"CMSURL",
"airwatchCertPath":"/path/to/cert",
"airwatchCertPass":"certPassword",
"airwatchAPIUser":"UserAPI",
"airwatchAPIPassword":"PasswordUserAPI",
"glpiServer":"http://127.0.0.1/glpi",
"glpiAppToken":"GLPIAppToken",
"glpiUserToken":"GLPIUserToken",
"stagingUser":"staging-pr",
"userAgent":"Airwatch Synchronizer"
}
settings = None
if(not os.path.isfile("./settings.json")):
f = open("./settings.json", "w")
f.write(json.dumps(settingsDefault, indent=4))
f.close()
exit(1)
# Récupération des informations du fichier de configuration
if(args.configpath != None and args.configpath != ''):
settings = getSettings(args.configpath)
else:
with open("./settings.json", "r") as f:
settings = json.load(f)
settings = getSettings("./conf/settings.conf")
#=========== Configuration des logs ===========#
logger = logging.getLogger(__name__)
if(args.debug or settings["LOGS"]["Debug"]):
logginglevel = logging.DEBUG
else:
logginglevel = logging.INFO
logger.setLevel(logginglevel)
formatter = logging.Formatter(fmt='%(asctime)s | %(levelname)s: %(message)s', datefmt='%Y/%m/%d %H:%M:%S')
# handler pour log dans un fichier
if(settings["LOGS"]["Enabled"]):
if(settings["LOGS"].get("Path") and settings["LOGS"].get("Path") != ""):
fileHandler = logging.FileHandler(f"{settings['LOGS'].get('Path')}syncGLPI.log")
else:
fileHandler = logging.FileHandler('./logs/syncGLPI.log')
fileHandler.setLevel(logginglevel)
fileHandler.setFormatter(formatter)
logger.addHandler(fileHandler)
# handler pour log dans la console
if(not args.silent):
consoleHandler = logging.StreamHandler()
consoleHandler.setLevel(logginglevel)
consoleHandler.setFormatter(formatter)
logger.addHandler(consoleHandler)
#======== Paramètres du script ========#
# Emplacement du verrou
lockFile = './airwatchSyncGLPI.lock'
debug=args.debug
logger.debug(f"============ Settings ============")
logger.debug(f"Airwatch server: {settings['AIRWATCH']['Server']}")
logger.debug(f"Authentication method : {settings['AIRWATCH']['AuthenticationMethod']}")
logger.debug(f"Staging user: {settings['AIRWATCH']['StagingUser']}")
logger.debug(f"GLPI server: {settings['GLPI']['Server']}")
logger.debug(f"UserAgent: {settings['GLPI']['UserAgent']}")
# Informations du serveur Airwatch
airwatchServer = settings["airwatchServer"]
airwatchAPIKey = settings["airwatchAPIKey"]
airwatchAuthMethod = settings["airwatchAuthMethod"]
airwatchAPIUser = None
airwatchAPIPassword = None
airwatchCertPath = None
airwatchCertPass = None
if(airwatchAuthMethod == 'password'):
airwatchAPIUser = settings["airwatchAPIUser"]
airwatchAPIPassword = settings["airwatchAPIPassword"]
else:
airwatchCertPath = settings["airwatchCertPath"]
airwatchCertPass = settings["airwatchCertPass"]
# Informations du serveur GLPI
GLPIServer = settings["glpiServer"]
GLPIAppToken = settings["glpiAppToken"]
GLPIUserToken = settings["glpiUserToken"]
# Filtres
searchFilter = args.searchFilter
searchValue = args.searchValue
searchFilter = args.searchfilter
searchValue = args.searchvalue
# Platform exclusion (12 = computer)
platformFilterEnabled = True
@ -80,100 +77,54 @@ platformFilterOut = [12]
# ====================================== #
def getAirwatchHeaders(airwatchAuthMethod, airwatchAPIKey, uri, User=None, password=None, CertPath=None, CertPassword=None):
if(airwatchAuthMethod == "password"):
airwatchAPIUserToken = base64.b64encode(f"{airwatchAPIUser}:{airwatchAPIPassword}".encode('ascii')).decode("ascii")
return {
"Authorization": f"Basic {airwatchAPIUserToken}",
"aw-tenant-code": airwatchAPIKey,
"Accept": "application/json"
}
else:
signing_data = uri.split('?')[0]
with open(CertPath, 'rb') as certfile:
cert = certfile.read()
key, certificate, additional_certs = pkcs12.load_key_and_certificates(cert, CertPassword.encode())
options = [pkcs7.PKCS7Options.DetachedSignature]
signed_data = pkcs7.PKCS7SignatureBuilder().set_data(signing_data.encode("UTF-8")).add_signer(certificate, key, hashes.SHA256()).sign(serialization.Encoding.DER, options)
signed_data_b64 = base64.b64encode(signed_data).decode()
return {
"Authorization": f"CMSURL'1 {signed_data_b64}",
"aw-tenant-code": airwatchAPIKey,
"Accept": "application/json"
}
# Vérification de la présence du verrou avant de continuer
if(os.path.isfile(lockFile) and not args.force):
if(debug):
print('Lock file is present, exiting...')
logger.debug('Lock file exists, exiting...')
exit(0)
else:
open(lockFile, "w").close()
logger.info("========= Synchronization started =========")
# Adresse de recherche des appareils
# avec limite de 500 appareils par page (limite max de l'API)
airwatchAPIDevicesSearchURI = f"/API/mdm/devices/search?pagesize=500&page="
airwatchHeaders = getAirwatchHeaders(airwatchAuthMethod, airwatchAPIKey, uri=airwatchAPIDevicesSearchURI, User=airwatchAPIUser, password=airwatchAPIPassword, CertPath=airwatchCertPath, CertPassword=airwatchCertPass)
# Page de départ pour la recherche
pageNumber = 0
# Initialisation de la variable devices qui va stocker l'ensemble des appareils trouvés
devices = []
uri = f"{airwatchServer}{airwatchAPIDevicesSearchURI}{pageNumber}"
if(debug):
print(f"Uri for device search on airwatch : {uri}")
result = requests.get(uri, headers=airwatchHeaders)
if(debug):
print(f"Result of request : {result}")
# On vérifie qu'on a bien un retour OK pour la requête API
if(result.status_code != 200):
print(result.json())
# Suppression du verrou
try:
airwatch = AirwatchAPI(settings)
airwatch.GetDevices()
logger.info("Airwatch server connection succeeded")
except Exception as error:
logger.critical(f"Connection to Airwatch server failed : {error}")
os.remove(lockFile)
exit(0)
exit(1)
result = result.json()
# Initialisation de l'api GLPI
try:
glpiapi = GLPIAPI(settings)
logger.info("GLPI server connection succeeded")
except requests.exceptions.ConnectionError as error:
logger.critical(f"Connection to GLPI server failed : {error}")
os.remove(lockFile)
exit(1)
# On fait une requête pour chaque page tant qu'on a pas atteint le nombre
# d'appareils trouvé dans la première requête
while(len(devices) != result["Total"]):
uri = f"{airwatchServer}{airwatchAPIDevicesSearchURI}{pageNumber}"
result = requests.get(uri, headers=airwatchHeaders).json()
devices += result["Devices"]
pageNumber += 1
# Recherche des appareils
devices = airwatch.GetDevices()
if(debug):
print(f"Nombre d'appareils {len(devices)}")
logger.info(f"Number of devices found in Airwatch : {len(devices)}")
# ====================== Début suppression des doublons ================================= #
# On récupére les numéros de série
serials = [device["SerialNumber"] for device in devices]
serials = [device.SerialNumber for device in devices]
# On garde ceux qui sont présent plus d'une fois et qui n'ont pas HUBNOSERIAL (BYOD) comme numéro de série
serials = {serial for serial in serials if serials.count(serial) > 1 and serial != 'HUBNOSERIAL'}
# Récupération des id et de la dernière date d'enrolement
# Récupération des devices et de la dernière date d'enrolement
devicesDouble = {}
for serial in serials:
# on fait une liste des appareils avec le même numéro de série
# que l'on stocke dans un dictionnaire avec le numéro de série en clé
devicesDouble[serial] = [[device["Id"]["Value"],datetime.strptime(device["LastEnrolledOn"], "%Y-%m-%dT%H:%M:%S.%f")] for device in devices if serial == device["SerialNumber"]]
devicesDouble[serial] = [[device,datetime.strptime(device.LastEnrolledOn, "%Y-%m-%dT%H:%M:%S.%f")] for device in devices if serial == device.SerialNumber]
if(debug):
print(f"Doublons détectés: {len(devicesDouble)}")
logger.info(f"Duplicates detected : {len(devicesDouble)}")
# On supprime les doublons qui ne se sont pas enrôlés en dernier
devicesToDelete = []
@ -189,220 +140,61 @@ for k,v in devicesDouble.items():
else:
devicesToDelete += [d[0]]
# On retire ces appareils de la liste
devices = [d for d in devices if d["Id"]["Value"] not in devicesToDelete]
# envoi de la requête de suppression des appareils sur magenta
airwatchAPIDeleteURI = '/API/mdm/devices/'
airwatchHeaders = getAirwatchHeaders(airwatchAuthMethod, airwatchAPIKey, uri=airwatchAPIDeleteURI, User=airwatchAPIUser, password=airwatchAPIPassword, CertPath=airwatchCertPath, CertPassword=airwatchCertPass)
# envoi de la requête de suppression des appareils sur airwatch
for device in devicesToDelete:
uri = f"{airwatchServer}{airwatchAPIDeleteURI}{device}"
if(debug):
print(f"Suppression de {uri}")
requests.delete(uri, headers=airwatchHeaders)
logger.info(f"Deleting {device.Id} - {device.FriendlyName} in Airwatch")
airwatch.DeleteDevice(device)
devices = airwatch.GetDevices()
# ====================== Fin suppression des doublons ================================= #
if(searchFilter != None):
if(debug):
print(f"SearchFilter set to {searchFilter}")
print(f"SearchValue set to {searchValue}")
logger.debug(f"SearchFilter set to {searchFilter}")
logger.debug(f"SearchValue set to {searchValue}")
if(searchFilter == 'Id'):
devices = [device for device in devices if device["Id"]["Value"] == searchValue]
devices = [device for device in devices if getattr(device, "Id") == searchValue]
else:
devices = [device for device in devices if device[searchFilter] == searchValue]
# Adresse d'initalisation de l'api GLPI
GLPIAPIInitUri = '/apirest.php/initSession/'
GLPIHeaders = {
'Content-Type': 'application/json',
"Authorization": f"user_token {GLPIUserToken}",
"App-Token": GLPIAppToken
}
# Récupération d'un token de session
uri = f"{GLPIServer}{GLPIAPIInitUri}"
result = requests.get(uri, headers=GLPIHeaders)
if(debug):
print(f"GLPI api access : {result}")
if(result.status_code != 200):
# Suppression du verrou
os.remove(lockFile)
exit(1)
GLPISessionToken = result.json()["session_token"]
if(debug):
print(f"GLPI session Token: {GLPISessionToken}")
# Changement des headers pour remplacer l'user token par le token de session
GLPIHeaders = {
'Content-Type': 'application/json',
"Session-Token": GLPISessionToken,
"App-Token": GLPIAppToken
}
# Adresse de recherche des appareils présents dans ordinateurs sur GLPI
GLPIAPISearchComputer = '/apirest.php/search/computer?'
platforms = {
2:"Apple iOS",
5:"Android",
12:"Windows Desktop"
}
processorArchs = {
0:{
"osArch":"arm64",
"softwareArch":"arm64"
},
9:{
"osArch":"64-bit",
"softwareArch":"x86_64"
}
}
devices = [device for device in devices if getattr(device, "searchFilter") == searchValue]
for device in devices:
if(device["EnrollmentStatus"] != 'Enrolled'):
if(device.EnrollmentStatus != 'Enrolled'):
logger.warning(f"Device with id {device.Id} not enrolled, skipping this device...")
continue
if(device["Imei"] != ''):
if(debug):
print(f"Imei = {device['Imei']}")
# Recherche des appareils en fonction du numéro de série ou de l'imei
# l'imei pouvant être dans le champ numéro de série ou les champs imei custom
search_parameter = f'is_deleted=0&criteria[0][field]=5&withindexes=true&criteria[0][searchtype]=contains&criteria[0][value]=^{device["SerialNumber"]}$'\
f'&criteria[1][link]=OR&criteria[1][field]=5&criteria[1][searchtype]=contains&criteria[1][value]=^{device["Imei"]}$'\
f'&criteria[2][link]=OR&criteria[2][field]=76667&criteria[2][searchtype]=contains&criteria[2][value]=^{device["Imei"]}$'\
f'&criteria[3][link]=OR&criteria[3][field]=76670&criteria[3][searchtype]=contains&criteria[3][value]=^{device["Imei"]}$'
else:
# Recherche des appareils en fonction du numéro de série seulement
search_parameter = f'is_deleted=0&criteria[0][field]=5&withindexes=true&criteria[0][searchtype]=contains&criteria[0][value]=^{device["SerialNumber"]}$'
if(debug):
print(f"Serial Number = {device['SerialNumber']}")
searchUri = f"{GLPIServer}{GLPIAPISearchComputer}{search_parameter}"
if(debug):
print(f"searchURI = {searchUri}")
search = requests.get(searchUri, headers=GLPIHeaders)
logger.info(f"Searching device {device.FriendlyName} (id={device.Id}) on GLPI")
# On ne gère pas pour l'instant d'autres code que le code 200
# voir en fonction des codes erreurs retournés par le serveur
if(search.status_code != 200):
break
deviceID, data, count = glpiapi.GetDevice(device)
apps = airwatch.GetDeviceApps(device)
if(count > 1):
logger.error(f"{count} devices matching airwatch device in GLPI (GLPI ids = {', '.join(deviceID)}), skipping this device...")
continue
if(count == 0):
logger.error(f"Device not found in GLPI, is it in the trash bin ? Skipping device...")
continue
search = search.json()
if(search["totalcount"] == 1):
# Récupération de l'utilisateur de l'appareil dans la fiche GLPI de l'appareil
for device_id, data in search["data"].items():
platformId = device["PlatformId"]["Id"]["Value"]
if(platformId in platforms.keys()):
platformName = platforms[platformId]
else:
platformName = "Unknown"
processorArch = device["ProcessorArchitecture"]
if(processorArch in processorArchs.keys()):
osArch = processorArchs[processorArch]["osArch"]
softwareArch = processorArchs[processorArch]["softwareArch"]
else:
osArch = "Unknown"
softwareArch = "Unknown"
inventory = {
"action":"inventory",
"content":{
"accesslog":{
"logdate": datetime.strptime(device["LastSeen"], "%Y-%m-%dT%H:%M:%S.%f").strftime("%Y-%m-%d %H:%M:%S")
},
"versionclient":settings["userAgent"],
"users":[
{
"login": device["UserName"]
}
],
"operatingsystem":{
"name": platformName,
"version": device["OperatingSystem"],
"full_name": f"{platformName} {device['OperatingSystem']}",
"arch": osArch
},
"softwares":[
],
"hardware":{
"name":data["1"],
"uuid":device["Uuid"],
"memory":device["TotalPhysicalMemory"]
}
},
"tag":device["LocationGroupName"],
"deviceid":f"{data['1']} - {device['SerialNumber']}",
"itemtype":"Computer"
}
# Récupération des applications présents sur les appareils
airwatchAPIAppsSearchURI = f"/api/mdm/devices/{device['Uuid']}/apps/search"
airwatchHeaders = getAirwatchHeaders(airwatchAuthMethod, airwatchAPIKey, uri=airwatchAPIAppsSearchURI, User=airwatchAPIUser, password=airwatchAPIPassword, CertPath=airwatchCertPath, CertPassword=airwatchCertPass)
uri = f"{airwatchServer}{airwatchAPIAppsSearchURI}"
apps = requests.get(uri, headers=airwatchHeaders).json()
for app in apps["app_items"]:
if(app["installed_status"] != "Installed"):
continue
install_date = datetime.strptime(app["latest_uem_action_time"], "%Y-%m-%dT%H:%M:%S.%f").strftime("%Y-%m-%d")
if(install_date == "1-01-01"):
inventory["content"]["softwares"] += [{
"name": app["name"],
"guid": app["bundle_id"],
"version": app["installed_version"],
"filesize": app["size"],
"arch": softwareArch
}]
else:
inventory["content"]["softwares"] += [{
"name": app["name"],
"guid": app["bundle_id"],
"version": app["installed_version"],
"install_date": install_date,
"filesize": app["size"],
"arch": softwareArch
}]
# Mise à jour du friendly name sur Airwatch
if(device["DeviceFriendlyName"] != f"{data['1']} {platformName} {device['OperatingSystem']} - {device['UserName']}"):
airwatchAPIURI = f"/API/mdm/devices/{device['Id']['Value']}"
airwatchHeaders = getAirwatchHeaders(airwatchAuthMethod, airwatchAPIKey, uri=airwatchAPIURI, User=airwatchAPIUser, password=airwatchAPIPassword, CertPath=airwatchCertPath, CertPassword=airwatchCertPass)
uri = f"{airwatchServer}{airwatchAPIURI}"
updateDeviceDetails = {
"DeviceFriendlyName":f"{data['1']} {platformName} {device['OperatingSystem']} - {device['UserName']}"
}
requests.put(uri, headers=airwatchHeaders, json=updateDeviceDetails)
inventory = glpiapi.CreateInventoryForAirwatchDevice(device, data["1"], apps)
# Mise à jour du friendly name sur Airwatch
platformName = inventory.operatingsystem["name"]
if(device.FriendlyName != f"{data['1']} {platformName} {device.OS} - {device.User}"):
newFriendlyName = f"{data['1']} {platformName} {device.OS} - {device.User}"
logger.info(f"Updating device friendlyname to {newFriendlyName}")
airwatch.SetDeviceFriendlyName(device, newFriendlyName)
headers = {
"Content-Type":"Application/x-compress",
"user-agent":settings["userAgent"]
}
if(debug):
print(f"Updating {device_id} on GLPI")
# filtre des plateformes
if(platformFilterEnabled):
if device["PlatformId"]["Id"]["Value"] in platformFilterOut:
continue
result = requests.post(GLPIServer, headers=headers, json=inventory)
if(debug):
print(result.json())
# filtre des plateformes
if(platformFilterEnabled):
if device.PlatformId in platformFilterOut:
logger.info(f"Device platform ({device.PlatformId}) is filtered out, not updating GLPI")
continue
logger.info(f"Updating {deviceID} on GLPI")
glpiapi.UpdateInventory(inventory.Json())
if(data['5'] != device.SerialNumber):
logger.info(f"Updating serial number from {data['5']} to {device.SerialNumber} in GLPI")
glpiapi.UpdateSerialNumber(deviceID, device.SerialNumber)
if(debug):
print('Removing lock')
logger.info("========= End of synchronization =========")
logger.debug('Removing lock file')
os.remove(lockFile)