405 lines
16 KiB
Python
405 lines
16 KiB
Python
import requests
|
|
import json
|
|
from datetime import datetime
|
|
|
|
class GLPIAPI:
|
|
def __init__(self, Server, AppToken, UserToken, UserAgent="GLPI API"):
|
|
self.Server = Server
|
|
self.AppToken = AppToken
|
|
self.UserToken = UserToken
|
|
self.UserAgent = 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 GetComputers(self, deviceName=None, serialNumber=None, user=None, imei=None, airwatchDevice=None):
|
|
'''
|
|
Search for computer items in GLPI based on one of the possibles parameters :
|
|
- deviceName : name of the item
|
|
- serialNumber : serial number of the item
|
|
- user : user of the item
|
|
- imei : custom field using field plugin (only for internal use)
|
|
- airwatchDevice : an airwatchDevice object from airwatchAPI module
|
|
If no parameters are set, it will search for all items
|
|
|
|
Return a tuple with item id, item data and item count.
|
|
'''
|
|
searchAll = False
|
|
if(deviceName != None):
|
|
# Recherche en fonction du nom de l'appareil
|
|
search_parameter = f'is_deleted=0&criteria[0][field]=1&withindexes=true&criteria[0][searchtype]=contains&criteria[0][value]=^{deviceName}$'
|
|
elif(user != None):
|
|
# Recherche en fonction de l'utilisateur de l'appareil
|
|
search_parameter = f'is_deleted=0&criteria[0][field]=70&withindexes=true&criteria[0][searchtype]=contains&criteria[0][value]=^{user}$'
|
|
elif(airwatchDevice != None):
|
|
if(airwatchDevice.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 (plugin field)
|
|
search_parameter = f'is_deleted=0&criteria[0][field]=5&withindexes=true&criteria[0][searchtype]=contains&criteria[0][value]=^{airwatchDevice.SerialNumber}$'\
|
|
f'&criteria[1][link]=OR&criteria[1][field]=5&criteria[1][searchtype]=contains&criteria[1][value]=^{airwatchDevice.Imei}$'\
|
|
f'&criteria[2][link]=OR&criteria[2][field]=76667&criteria[2][searchtype]=contains&criteria[2][value]=^{airwatchDevice.Imei}$'\
|
|
f'&criteria[3][link]=OR&criteria[3][field]=76670&criteria[3][searchtype]=contains&criteria[3][value]=^{airwatchDevice.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]=^{airwatchDevice.SerialNumber}$'
|
|
|
|
elif(imei != None):
|
|
# 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 (plugin field)
|
|
search_parameter = f'is_deleted=0&criteria[0][field]=5&withindexes=true&criteria[0][searchtype]=contains&criteria[0][value]=^{serialNumber}$'\
|
|
f'&criteria[1][link]=OR&criteria[1][field]=5&criteria[1][searchtype]=contains&criteria[1][value]=^{imei}$'\
|
|
f'&criteria[2][link]=OR&criteria[2][field]=76667&criteria[2][searchtype]=contains&criteria[2][value]=^{imei}$'\
|
|
f'&criteria[3][link]=OR&criteria[3][field]=76670&criteria[3][searchtype]=contains&criteria[3][value]=^{imei}$'
|
|
|
|
elif(serialNumber != None):
|
|
# 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]=^{serialNumber}$'
|
|
|
|
else:
|
|
# Recherche de l'ensemble des appareils si rien n'est renseigné
|
|
searchAll = True
|
|
|
|
search_parameter = f'is_deleted=0&criteria[0][link]=AND&criteria[0][field]=view&criteria[0][searchtype]=contains&criteria[0][value]='
|
|
if(searchAll):
|
|
searchUri = f"{self.Server}/apirest.php/search/Computer/?range=0-9999999&is_deleted=0"
|
|
else:
|
|
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):
|
|
itemID = list(search["data"].keys())[0]
|
|
data = search["data"][itemID]
|
|
|
|
return itemID, data, search["totalcount"]
|
|
elif(search["totalcount"] > 1):
|
|
if(searchAll):
|
|
# requires id to be in the display preferences of the api user
|
|
itemID = [i["2"] for i in search["data"]]
|
|
else:
|
|
itemID = list(search["data"].keys())
|
|
return itemID, search["data"], search["totalcount"]
|
|
|
|
return None, None, 0
|
|
|
|
def GetUsers(self, username=None, email=None):
|
|
'''
|
|
Search for users in GLPI based on one of the possibles parameters :
|
|
- username : username of the glpi user
|
|
- email : email of the glpi user
|
|
If no parameters are set, it will search for all users
|
|
|
|
Returns a tuple with user id, user data and user count
|
|
'''
|
|
searchAll = False
|
|
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}$'
|
|
else:
|
|
searchAll = True
|
|
|
|
if(searchAll):
|
|
searchUri = f"{self.Server}/apirest.php/search/User/?range=0-9999999&is_deleted=0"
|
|
else:
|
|
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):
|
|
if(searchAll):
|
|
# requires id to be in the display preferences of the api user
|
|
userID = [i["2"] for i in search["data"]]
|
|
else:
|
|
userID = list(search["data"].keys())
|
|
return userID, search["data"], search["totalcount"]
|
|
|
|
return None, None, 0
|
|
|
|
def UploadFile(self, file, path):
|
|
manifest = {
|
|
"input": {
|
|
"name": file,
|
|
"is_recursive": 1,
|
|
"_filename": [file],
|
|
"documentcategories_id": 4
|
|
}
|
|
}
|
|
fileData = open(path+file, "rb")
|
|
headers = {
|
|
"Session-Token": self.SessionToken,
|
|
"App-Token": self.AppToken
|
|
}
|
|
data = {"uploadManifest": json.dumps(manifest)}
|
|
files = {file: fileData}
|
|
return requests.post(self.Server+"/apirest.php/Document/", headers=headers, data=data, files=files)
|
|
|
|
def SetDocumentToItem(self, itemID, itemType, documentID):
|
|
body = {
|
|
"input": {
|
|
"documents_id": documentID,
|
|
"items_id": itemID,
|
|
"itemtype": itemType
|
|
}
|
|
}
|
|
return requests.post(self.Server+"/apirest.php/Document/"+str(documentID)+"/Document_Item", headers=self.Headers, json=body)
|
|
|
|
def UpdateInventory(self, inventory):
|
|
headers = {
|
|
"Content-Type":"Application/x-compress",
|
|
"user-agent":self.UserAgent
|
|
}
|
|
return requests.post(self.Server, headers=headers, data=inventory)
|
|
|
|
def SetField(self, itemType, itemId, data):
|
|
'''Modify fields of a GLPI item.
|
|
Input must be a dict formatted like this :
|
|
{
|
|
"fieldName1": "fieldValue1",
|
|
"fieldName2": "fieldValue2",
|
|
...
|
|
}
|
|
'''
|
|
body = {
|
|
"input" : {
|
|
"id" : itemId,
|
|
}
|
|
}
|
|
body["input"] = {**body["input"], **data}
|
|
|
|
uri = f"{self.Server}/apirest.php/{itemType}/"
|
|
return requests.put(uri, headers=self.Headers, json=body)
|
|
|
|
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 UpdateItemUser(self, itemID, itemType, username):
|
|
|
|
body = {
|
|
"input" : {
|
|
"id" : itemID,
|
|
"users_id" : username
|
|
}
|
|
}
|
|
uri = f"{self.Server}/apirest.php/{itemType}/"
|
|
return requests.put(uri, headers=self.Headers, json=body)
|
|
|
|
def SetCustomField(self, itemType, containerName, containerID, itemId, fieldName, data):
|
|
'''Requires fields plugin on GLPI server
|
|
- containerName is block label name
|
|
- containerID is block id
|
|
'''
|
|
uri = f"{self.Server}/apirest.php/PluginFields{itemType}{containerName}"
|
|
searchURI = f"{self.Server}/apirest.php/PluginFields{itemType}{containerName}?range=0-999999999"
|
|
result = requests.get(searchURI, headers=self.Headers)
|
|
if(result.status_code == 200):
|
|
result = result.json()
|
|
fieldItem = None
|
|
# searching for field item
|
|
for entry in result:
|
|
if str(entry["items_id"]) == itemId:
|
|
fieldItem = entry
|
|
|
|
if(fieldItem == None):
|
|
body = {
|
|
"input": {
|
|
"items_id": itemId,
|
|
"itemtype": itemType,
|
|
"plugin_fields_containers_id": containerID,
|
|
fieldName: data
|
|
}
|
|
}
|
|
return requests.post(uri, headers=self.Headers, json=body)
|
|
else:
|
|
body = {
|
|
"input": {
|
|
"id": fieldItem["id"],
|
|
fieldName: data
|
|
}
|
|
}
|
|
return requests.put(uri, headers=self.Headers, json=body)
|
|
|
|
def CreateInventoryForAirwatchDevice(self, device, deviceName, apps=None):
|
|
platforms = {
|
|
2:"Apple iOS",
|
|
5:"Android",
|
|
10:"Apple macOS",
|
|
12:"Windows"
|
|
}
|
|
|
|
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"
|
|
|
|
windowsOSTranslation = {
|
|
"10.0.19043":"10 21H1",
|
|
"10.0.19044":"10 21H2",
|
|
"10.0.19045":"10 22H2",
|
|
"10.0.22000":"11 21H2",
|
|
"10.0.22621":"11 22H2",
|
|
"10.0.22631":"11 23H2",
|
|
"10.0.26100":"11 24H2",
|
|
"10.0.26200":"11 25H2"
|
|
}
|
|
|
|
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")
|
|
if(platformName == "Windows"):
|
|
if(device.OS in windowsOSTranslation.keys()):
|
|
inventory.SetOperatingSystem(platformName, windowsOSTranslation[str(device.OS)], osArch)
|
|
else:
|
|
platformName = "Windows Desktop"
|
|
inventory.SetOperatingSystem(platformName, device.OS, osArch)
|
|
else:
|
|
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)
|