refactor rebase

This commit is contained in:
Cayo Puigdefabregas 2024-12-17 08:43:39 +01:00
parent 4d2905799f
commit 82e60dff24

View file

@ -28,23 +28,6 @@ SNAPSHOT_BASE = {
} }
## Legacy Functions ##
def convert_to_legacy_snapshot(snapshot):
snapshot["sid"] = str(uuid.uuid4()).split("-")[1]
snapshot["software"] = "workbench-script"
snapshot["version"] = "dev"
snapshot["schema_api"] = "1.0.0"
snapshot["settings_version"] = "No Settings Version (NaN)"
snapshot["timestamp"] = snapshot["timestamp"].replace(" ", "T")
snapshot["data"]["smart"] = snapshot["data"]["disks"]
snapshot["data"]["lshw"] = json.loads(snapshot["data"]["lshw"])
snapshot["data"].pop("disks")
snapshot.pop("erase")
snapshot.pop("token_hash")
## End Legacy Functions ##
## Utility Functions ## ## Utility Functions ##
def logs(f): def logs(f):
@ -73,14 +56,7 @@ def exec_cmd_erase(cmd):
## End Utility functions ## ## End Utility functions ##
def convert_to_credential(snapshot): ## Legacy Functions ##
snapshot["data"] = json.dumps(snapshot["data"])
file_path = os.path.join(BASE_DIR, "templates", "snapshot.json")
with open(file_path) as f:
ff = f.read()
template = Template(ff)
cred = template.substitute(**snapshot)
return cred
def convert_to_legacy_snapshot(snapshot): def convert_to_legacy_snapshot(snapshot):
snapshot["sid"] = str(uuid.uuid4()).split("-")[1] snapshot["sid"] = str(uuid.uuid4()).split("-")[1]
@ -417,62 +393,6 @@ def send_snapshot_to_devicehub(snapshot, token, url, ev_uuid, legacy):
except Exception as e: except Exception as e:
logger.error(_("Snapshot not remotely sent to URL '%s'. Do you have internet? Is your server up & running? Is the url token authorized?\n %s"), url, e) logger.error(_("Snapshot not remotely sent to URL '%s'. Do you have internet? Is your server up & running? Is the url token authorized?\n %s"), url, e)
except Exception as e:
logger.error(_("Credential not remotely builded to URL '%s'. Do you have internet? Is your server up & running? Is the url token authorized?\n %s"), url, e)
return json.dumps(snapshot)
# TODO sanitize url, if url is like this, it fails
# url = 'http://127.0.0.1:8000/api/snapshot/'
def send_snapshot_to_devicehub(snapshot, token, url, ev_uuid):
url_components = urllib.parse.urlparse(url)
ev_path = f"evidence/{ev_uuid}"
components = (url_components.scheme, url_components.netloc, ev_path, '', '', '')
ev_url = urllib.parse.urlunparse(components)
# apt install qrencode
qr = "echo {} | qrencode -t ANSI".format(ev_url)
print(exec_cmd(qr))
print(ev_url)
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
try:
data = snapshot.encode('utf-8')
request = urllib.request.Request(url, data=data, headers=headers)
with urllib.request.urlopen(request) as response:
status_code = response.getcode()
#response_text = response.read().decode('utf-8')
if 200 <= status_code < 300:
logger.info(_("Snapshot successfully sent to '%s'"), url)
if legacy:
try:
response = json.loads(response_text)
public_url = response.get('public_url')
dhid = response.get('dhid')
if public_url:
# apt install qrencode
qr = "echo {} | qrencode -t ANSI".format(public_url)
print(exec_cmd(qr))
print("url: {}".format(public_url))
if dhid:
print("dhid: {}".format(dhid))
except Exception:
logger.error(response_text)
else:
qr = "echo {} | qrencode -t ANSI".format(ev_url)
print(exec_cmd(qr))
print(f"url: {ev_url}")
else:
logger.error(_("Snapshot cannot sent to '%s'"), url)
except Exception as e:
logger.error(_("Snapshot not remotely sent to URL '%s'. Do you have internet? Is your server up & running? Is the url token authorized?\n %s"), url, e)
def load_config(config_file="settings.ini"): def load_config(config_file="settings.ini"):
""" """