Files
glpi-airwatch-sync/scripts/includes/GLPIAPI.py
2025-09-12 12:48:03 +02:00

244 lines
9.1 KiB
Python
Executable File

import requests
import json
from datetime import datetime
class GLPIAPI:
def __init__(self, settings):
self.Server = settings["GLPI"]["Server"]
self.AppToken = settings["GLPI"]["AppToken"]
self.UserToken = settings["GLPI"]["UserToken"]
self.UserAgent = settings["GLPI"]["UserAgent"]
self.SessionToken = None
self.StatusCode = None
self.Headers = None
self.InitConnection()
def InitConnection(self):
initURI = '/apirest.php/initSession/'
GLPIHeaders = {
'Content-Type': 'application/json',
"Authorization": f"user_token {self.UserToken}",
"App-Token": self.AppToken
}
# Récupération d'un token de session
uri = f"{self.Server}{initURI}"
result = requests.get(uri, headers=GLPIHeaders)
self.StatusCode = result.status_code
if(result.status_code == 200):
self.SessionToken = result.json()["session_token"]
self.Headers = {
'Content-Type': 'application/json',
"Session-Token": self.SessionToken,
"App-Token": self.AppToken
}
def GetDevice(self, device):
if(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}$'
searchUri = f"{self.Server}/apirest.php/search/computer?{search_parameter}"
search = requests.get(searchUri, headers=self.Headers)
if(search.status_code == 200):
search = search.json()
if(search["totalcount"] == 1):
deviceID = list(search["data"].keys())[0]
data = search["data"][deviceID]
return deviceID, data, search["totalcount"]
elif(search["totalcount"] > 1):
deviceID = list(search["data"].keys())
return deviceID, search["data"], search["totalcount"]
else:
return None, None, 0
return None, None, None
def GetUser(self, username=None, email=None):
if(username != None):
search_parameter = f'is_deleted=0&criteria[0][field]=1&withindexes=true&criteria[0][searchtype]=contains&criteria[0][value]=^{username}$'
elif(email != None):
search_parameter = f'is_deleted=0&criteria[0][field]=5&withindexes=true&criteria[0][searchtype]=contains&criteria[0][value]=^{email}$'
searchUri = f"{self.Server}/apirest.php/search/user?{search_parameter}"
search = requests.get(searchUri, headers=self.Headers)
if(search.status_code == 200):
search = search.json()
if(search["totalcount"] == 1):
userID = list(search["data"].keys())[0]
data = search["data"][userID]
return userID, data, search["totalcount"]
elif(search["totalcount"] > 1):
userID = list(search["data"].keys())
return userID, search["data"], search["totalcount"]
else:
return None, None, 0
def UpdateInventory(self, inventory):
headers = {
"Content-Type":"Application/x-compress",
"user-agent":self.UserAgent
}
return requests.post(self.Server, headers=headers, data=inventory)
def UpdateSerialNumber(self, deviceid, serialnumber):
body = {
"input" : {
"id" : deviceid,
"serial" : serialnumber
}
}
uri = f"{self.Server}/apirest.php/Computer/"
return requests.put(uri, headers=self.Headers, json=body)
def UpdateUser(self, deviceid, username):
body = {
"input" : {
"id" : deviceid,
"users_id" : username
}
}
uri = f"{self.Server}/apirest.php/Computer/"
return requests.put(uri, headers=self.Headers, json=body)
def CreateInventoryForAirwatchDevice(self, device, deviceName, apps=None):
platforms = {
2:"Apple iOS",
5:"Android",
12:"Windows Desktop"
}
if(device.PlatformId in platforms.keys()):
platformName = platforms[device.PlatformId]
else:
platformName = "Unknown"
processorArchs = {
0:{
"osArch":"arm64",
"softwareArch":"arm64"
},
9:{
"osArch":"64-bit",
"softwareArch":"x86_64"
}
}
if(device.Arch in processorArchs.keys()):
osArch = processorArchs[device.Arch]["osArch"]
softwareArch = processorArchs[device.Arch]["softwareArch"]
else:
osArch = "Unknown"
softwareArch = "Unknown"
logDate = datetime.strptime(device.LastSeen, "%Y-%m-%dT%H:%M:%S.%f").strftime("%Y-%m-%d %H:%M:%S")
inventory = GLPIInventory(logdate=logDate, versionclient=self.UserAgent, tag=device.Group, deviceid=f"{deviceName} - {device.SerialNumber}", itemtype="Computer")
inventory.SetOperatingSystem(platformName, device.OS, osArch)
inventory.SetHardware(deviceName, device.Uuid, device.TotalMemory)
inventory.AddUser(device.User)
if(apps != None):
for app in apps:
if(app.Status != "Installed"):
continue
install_date = datetime.strptime(app.InstallDate, "%Y-%m-%dT%H:%M:%S.%f").strftime("%Y-%m-%d")
if(install_date == "1-01-01"):
inventory.AddSoftware(app.Name, app.Version, app.Size, softwareArch, app.Guid)
else:
inventory.AddSoftware(app.Name, app.Version, app.Size, softwareArch, app.Guid, install_date)
return inventory
class GLPIInventory:
def __init__(self, logdate=None, versionclient=None, tag=None, deviceid=None, itemtype=None):
self.logdate = logdate
self.versionclient = versionclient
self.users = []
self.operatingsystem = {}
self.softwares = []
self.hardware = {}
self.tag = tag
self.deviceId = deviceid
self.itemType = itemtype
def AddUser(self, user):
self.users += [{
"login": user
}]
def DelUser(self, user):
for i in range(0, len(self.users)):
if(self.users[i]["login"] == user):
del self.users[i]
def AddSoftware(self, name, version, filesize, arch, guid, install_date=None):
if(install_date == None):
self.softwares += [{
"name": name,
"guid": guid,
"version": version,
"filesize": filesize,
"arch": arch
}]
else:
self.softwares += [{
"name": name,
"guid": guid,
"version": version,
"install_date": install_date,
"filesize": filesize,
"arch": arch
}]
def DelSoftware(self, guid):
for i in range(0, len(self.softwares)):
if(self.softwares[i]["guid"] == guid):
del self.softwares[i]
def SetHardware(self, name, uuid, memory):
self.hardware = {
"name": name,
"uuid": uuid,
"memory": memory
}
def SetOperatingSystem(self, name, version, arch):
self.operatingsystem = {
"name": name,
"version": version,
"full_name": f"{name} {version}",
"arch": arch
}
def Json(self):
inventory = {
"action": "inventory",
"content":{
"accesslog":{
"logdate": self.logdate
},
"versionclient": self.versionclient,
"users": self.users,
"operatingsystem": self.operatingsystem,
"softwares": self.softwares,
"hardware": self.hardware
},
"tag": self.tag,
"deviceid": self.deviceId,
"itemtype": self.itemType
}
return json.dumps(inventory)