puffb/puffb.py
2024-06-23 22:44:36 -04:00

494 lines
16 KiB
Python

"""Pretty simple Python module that makes interacting with a Pufferpanel Server Daemon easy."""
import requests
import json
from time import time, sleep
class ServerError(Exception):
pass
class Panel:
"""
A Panel object must get assigned 3 values: serverURL, clientID, and clientSecret.
These can be obtained by creating an oauth2 client in PufferPanel.
The user can also (optionally) define serverID, which is useful if the scope of your token doesn't cover multiple servers.
When serverID is defined, the serverID parameter can be skipped in methods where it is required, so
object.logs('a1b236')
becomes
object.logs()
"""
def __init__(self, serverUrl: str, clientId: str, clientSecret: str, serverId=None):
self.URL = serverUrl
self.ID = clientId
self.SECRET = clientSecret
self.SERVER_ID = serverId
if self.SERVER_ID:
try:
self.name = self.api_server()["server"]["name"]
except KeyError:
self.name = "Unknown"
try:
self.type = self.api_server()["server"]["type"]
except KeyError:
self.type = "generic-type"
else:
if not (self.getToken()):
raise ValueError(
"Could not obtain a token from the server. Are your credentials correct?"
)
def __str__(self):
return (
f'URL: {self.URL}, Client ID: {self.ID}, using "{self.name}" ({self.SERVER_ID}) running a {self.type} server'
if self.SERVER_ID
else f"URL: {self.URL}, Client ID: {self.ID}"
)
# What this might look like:
# "URL: https://panel.domain.com, Client ID: randomClientId, using YourServer (randomSecretKey) running a type-of server"
#
# or
#
# "URL: https://panel.domain.com, Client ID: randomClientId"
def __repr__(self):
return (
f"Panel('{self.URL}', '{self.ID}', '{self.SECRET}', '{self.SERVER_ID}')"
if self.SERVER_ID
else f"Panel('{self.URL}', '{self.ID}', '{self.SECRET}')"
)
# What this might look like:
# Panel('https://panel.domain.com', 'randomClientId', 'randomSecretKey', 'a1b236')
#
# or
#
# Panel('https://panel.domain.com', 'randomClientId', 'randomSecretKey')
def invalidCode(self, given: requests.Response, override=False) -> bool:
validCodes = []
for i in range(200, 300):
validCodes.append(i)
isInvalidCode = not (given.status_code in validCodes)
isServerOfflineMsg = given.json() == {
"error": {"msg": "server offline", "code": "ErrServerOffline"}
}
return (
isInvalidCode or isServerOfflineMsg if not (override) else isInvalidCode
) # Valid codes are 200-299.
# The server can also return an 'error' as written here, even if the request was successful.
# I've just made it so that it behaves exactly as if it had gotten an invalid response code, but added a manual override so that creating an object while this special error is returned is still possible.
# If the response message changes or if there is a better way to handle this I'll change it
def getToken(self) -> str:
"""
Obtains a token from the server with the provided client ID and secret.
Returns the token if succesful, error if not.
This function is called anytime there is any contact with the daemon,
because the token expires after a certain amount of time.
"""
GET_TOKEN_HEADER = {"Content-Type": "application/x-www-form-urlencoded"}
response = requests.post(
f"{self.URL}/oauth2/token",
data=f"grant_type=client_credentials&client_id={self.ID}&client_secret={self.SECRET}",
headers=GET_TOKEN_HEADER,
)
if self.invalidCode(response):
print(self.invalidCode(response))
raise ValueError(
"Could not obtain a token from the server. Are your credentials corret?"
)
accessToken = response.json()["access_token"]
return accessToken
def tokenIt(self, serverID) -> tuple:
token = self.getToken()
serverID = serverID or self.SERVER_ID
if not (serverID):
raise TypeError("Missing 1 required argument: 'serverID'")
AUTH_HEADER = {"Authorization": f"Bearer {token}"}
return (
serverID,
AUTH_HEADER,
) # tuple[0] is the server ID and tuple[1] is the HTTP header
# All GET methods
def api_server(self, serverID=None) -> dict:
"""
Gets the data of a given server from the API, not the config.
Returns the physical data of the server.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/api/servers/{data[0]}"
response = requests.get(fullURL, headers=data[1])
if self.invalidCode(response):
raise ServerError(f"Server returned an invalid response: {response.json()}")
return response.json()
# In it's current state this method was only implemented to get the server name, not much else.
def daemon(self) -> bool:
"""
Check if the API is active.
Different from status(), which checks if the given server is running, not the actual API.
Returns True if running, False if not.
"""
data = {"Authorization": f"Bearer {self.getToken()}"}
fullURL = f"{self.URL}/daemon"
response = requests.get(fullURL, headers=data)
if self.invalidCode(response):
return False
isRunning = response.json()["message"] == "daemon is running"
# Returns True if the server is running ASSUMING that the running message is "daemon is running". If this message changes I'll fix it
return isRunning
def data_admin(self, serverID=None) -> dict:
"""
Gets the full data of a server given a server ID as admin.
Returns the data of the server.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}"
response = requests.get(fullURL, headers=data[1])
if self.invalidCode(response):
raise ServerError(f"Server returned an invalid response: {response.json()}")
if response.json() == {"data": {}}:
raise ServerError(
f"Server returned an empty dataset: {response.json()}\nAre you using a client that inherits permissions for all servers?"
)
return response.json()["data"]
# All server data is within one key: 'data'. If this is a bad idea I'll change it
def logs(self, serverID=None) -> str:
"""
Gets the server log given a server ID with the process described at
https://hosting.povario.com/swagger/index.html#/default/get_daemon_server__id__console
This is different from the client ID, which is only used when obtaining a token with getToken().
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/console" # Todo: Implement selecting a timestamp
response = requests.get(fullURL, headers=data[1])
response.raise_for_status()
if self.invalidCode(response, override=True):
raise ServerError(f"Server returned an invalid response: {response.json()}")
logs = response.json()["logs"]
return logs
def data(self, serverID=None) -> dict:
"""
Gets the full data of a server given a server ID with the process described at
https://hosting.povario.com/swagger/index.html#/default/get_daemon_server__id_
Returns the data of the server.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/data"
response = requests.get(fullURL, headers=data[1])
if self.invalidCode(response):
raise ServerError(f"Server returned an invalid response: {response.json()}")
if response.json() == {"data": {}}:
raise ServerError(
f"Server returned an empty dataset: {response.json()}\nAre you using a client that inherits permissions for all servers?"
)
return response.json()["data"] # Same as data_admin().
def extract(self, filename: str, serverID=None) -> int:
"""
Extracts a given file on the given server.
Only returns the HTTP response code.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/extract/{filename}"
response = requests.get(fullURL, headers=data[1])
if self.invalidCode(response):
raise ServerError(f"Server returned an invalid response: {response.json()}")
return response.status_code
# def get_file(self,filename,serverID=None):
# """
# Lists a file or directory on the given server.
#
# Returns either a dictonary with the file properties or a dictionary with individual files in a directory.
# """
#
# data=self.tokenIt(serverID)
# fullURL=f'{self.URL}/proxy/daemon/server/{data[0]}/file/{filename}'
#
# response=requests.get(f'{fullURL}',
# headers=data[1])
#
# if self.invalidCode(response):
# raise ServerError(f'Server returned an invalid response: {response.json()}')
#
# return response.json()
#
# Kinda broken. Don't use please thank you
def stats(self, serverID=None, precise=False) -> dict:
"""
Gets the server stats given a server ID with the process described at
https://hosting.povario.com/swagger/index.html#/default/get_daemon_server__id__stats
Returns a dictonary with the following values rounded to two decimal places:
CPU Usage (float)
RAM Usage in MB (float)
Unless precise=True, in which case the values will not be rounded.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/stats"
response = requests.get(fullURL, headers=data[1])
if self.invalidCode(response):
return response.json()
cpu = response.json()["cpu"]
ram = (response.json()["memory"] / 1000) / 1000
formattedUsage = (
{"cpu_usage": round(cpu, 2), "ram_usage": round(ram, 2)}
if not (precise)
else {"cpu_usage": cpu, "ram_usage": ram}
)
return formattedUsage
def status(self, serverID=None) -> str:
"""
Checks if the server is running.
Returns True if running, False if not.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/status"
response = requests.get(fullURL, headers=data[1])
if self.invalidCode(response):
raise ServerError(f"Server returned an invalid response: {response.json()}")
return response.json()["running"]
# All POST methods
def update(self, serverID=None) -> int:
"""
Updates a server.
Only returns the HTTP response code.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}"
response = requests.post(fullURL, headers=data[1])
return response.status_code
def archive(self, filename, serverID=None) -> int:
"""
Archives a given file on the server.
Only returns the HTTP response code.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/archive/{filename}"
response = requests.post(fullURL, data=f"{filename}", headers=data[1])
return response.status_code
def exec(self, command, serverID=None) -> int:
"""
Executes a command on a given server.
Only returns the HTTP response code.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/console"
response = requests.post(fullURL, data=command, headers=data[1])
# previousLogs = self.logs(serverID)
# reply=self.logs(serverID)
# return reply.replace(previousLogs,'')
#
# Todo: return the logs instead of just the status code.
return response.status_code
def edit_data(self, key, value, serverID=None) -> int:
"""
Edits the server config.
Only returns the HTTP Response code.
"""
data = self.tokenIt(serverID)
reply = self.data(serverID)
options = []
for option in reply:
options.append(option)
optionExists = key in options
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/data"
if optionExists:
nonMatchingValue = type(reply[key]["value"]) != type(value)
if nonMatchingValue:
raise ValueError(
f"Value must match type for key '{key}'\n{type(value)} does not match {type(reply[key]['value'])}"
)
else:
raise ValueError(
f"'{key}' does not match any in the options list:\n{options}"
)
reply[key]["value"] = value
response = requests.post(
fullURL, data=json.dumps({"data": reply}), headers=data[1]
)
return response.status_code
def install(self, serverID=None) -> int:
"""
Queues an installation for the given server.
Potentially destructive action. (Can overwrite needed data such as config files)
Only returns the HTTP response code.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/install"
response = requests.post(fullURL, headers=data[1])
return response.status_code
def kill(self, serverID=None) -> int:
"""
Force quits a server without warning.
Only returns the HTTP response code.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/kill"
response = requests.post(fullURL, headers=data[1])
return response.status_code
def reload(self, serverID=None) -> int:
"""
Reloads a server from disk.
Potentially destructive action. (May overwrite server data)
This method is NOT the same as restart(), which stops then starts a server.
Only returns the HTTP response code.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/reload"
response = requests.post(fullURL, headers=data[1])
return response.status_code
def start(self, serverID=None):
"""
Starts the given server.
Only returns the HTTP response code.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/start"
response = requests.post(fullURL, headers=data[1])
return response.status_code
def stop(self, serverID=None):
"""
Stops the given server.
Only returns the HTTP response code.
"""
data = self.tokenIt(serverID)
fullURL = f"{self.URL}/proxy/daemon/server/{data[0]}/stop"
response = requests.post(fullURL, headers=data[1])
return response.status_code
def restart(self, serverID=None) -> int:
"""
Stops, then starts a given server.
Only returns the HTTP status code from the start() method.
"""
self.stop(serverID)
sleep(1.5)
response = self.start(serverID)
return response