Some changes to Utilities folder
This commit is contained in:
Binary file not shown.
Binary file not shown.
Executable
+345
@@ -0,0 +1,345 @@
|
||||
#!/usr/bin/env python
|
||||
# 0.0.0
|
||||
from Scripts import *
|
||||
import os, tempfile, datetime, shutil, time, plistlib, json, sys, argparse
|
||||
|
||||
class MountEFI:
|
||||
def __init__(self, **kwargs):
|
||||
self.r = run.Run()
|
||||
self.d = disk.Disk()
|
||||
self.dl = downloader.Downloader()
|
||||
self.u = utils.Utils("MountEFI")
|
||||
self.re = reveal.Reveal()
|
||||
# Get the tools we need
|
||||
self.script_folder = "Scripts"
|
||||
self.update_url = "https://raw.githubusercontent.com/corpnewt/MountEFIv2/master/MountEFI.command"
|
||||
|
||||
self.settings_file = kwargs.get("settings", None)
|
||||
cwd = os.getcwd()
|
||||
os.chdir(os.path.dirname(os.path.realpath(__file__)))
|
||||
if self.settings_file and os.path.exists(self.settings_file):
|
||||
self.settings = json.load(open(self.settings_file))
|
||||
else:
|
||||
self.settings = {
|
||||
# Default settings here
|
||||
"default_disk" : None,
|
||||
"after_mount" : None,
|
||||
"full_layout" : False,
|
||||
"skip_countdown" : False,
|
||||
}
|
||||
os.chdir(cwd)
|
||||
self.full = self.settings.get("full_layout", False)
|
||||
|
||||
def check_update(self):
|
||||
# Checks against https://raw.githubusercontent.com/corpnewt/MountEFIv2/master/MountEFI.command to see if we need to update
|
||||
self.u.head("Checking for Updates")
|
||||
print(" ")
|
||||
with open(os.path.realpath(__file__), "r") as f:
|
||||
# Our version should always be the second line
|
||||
version = get_version(f.read())
|
||||
print(version)
|
||||
try:
|
||||
new_text = _get_string(url)
|
||||
new_version = get_version(new_text)
|
||||
except:
|
||||
# Not valid json data
|
||||
print("Error checking for updates (network issue)")
|
||||
return
|
||||
|
||||
if version == new_version:
|
||||
# The same - return
|
||||
print("v{} is already current.".format(version))
|
||||
return
|
||||
# Split the version number
|
||||
try:
|
||||
v = version.split(".")
|
||||
cv = new_version.split(".")
|
||||
except:
|
||||
# not formatted right - bail
|
||||
print("Error checking for updates (version string malformed)")
|
||||
return
|
||||
|
||||
if not need_update(cv, v):
|
||||
print("v{} is already current.".format(version))
|
||||
return
|
||||
|
||||
# Update
|
||||
with open(os.path.realpath(__file__), "w") as f:
|
||||
f.write(new_text)
|
||||
|
||||
# chmod +x, then restart
|
||||
run_command(["chmod", "+x", __file__])
|
||||
os.execv(__file__, sys.argv)
|
||||
|
||||
def flush_settings(self):
|
||||
if self.settings_file:
|
||||
cwd = os.getcwd()
|
||||
os.chdir(os.path.dirname(os.path.realpath(__file__)))
|
||||
json.dump(self.settings, open(self.settings_file, "w"))
|
||||
os.chdir(cwd)
|
||||
|
||||
def after_mount(self):
|
||||
self.u.resize(80, 24)
|
||||
self.u.head("After Mount Action")
|
||||
print(" ")
|
||||
print("1. Return to Menu")
|
||||
print("2. Quit")
|
||||
print("3. Open EFI and Return to Menu")
|
||||
print("4. Open EFI and Quit")
|
||||
if not self.settings.get("skip_countdown", False):
|
||||
print("5. Skip After-Mount Countdown")
|
||||
else:
|
||||
print("5. Use After-Mount Countdown")
|
||||
print(" ")
|
||||
print("M. Main Menu")
|
||||
print("Q. Quit")
|
||||
print(" ")
|
||||
menu = self.u.grab("Please pick an option: ")
|
||||
if not len(menu):
|
||||
self.after_mount()
|
||||
return
|
||||
menu = menu.lower()
|
||||
if menu in ["1","2","3","4"]:
|
||||
self.settings["after_mount"] = [
|
||||
"Return to Menu",
|
||||
"Quit",
|
||||
"Reveal and Return to Menu",
|
||||
"Reveal and Quit"
|
||||
][int(menu)-1]
|
||||
self.flush_settings()
|
||||
return
|
||||
elif menu == "5":
|
||||
cd = self.settings.get("skip_countdown", False)
|
||||
cd ^= True
|
||||
self.settings["skip_countdown"] = cd
|
||||
self.flush_settings()
|
||||
self.after_mount()
|
||||
return
|
||||
elif menu == "m":
|
||||
return
|
||||
elif menu == "q":
|
||||
self.u.custom_quit()
|
||||
self.after_mount()
|
||||
|
||||
def default_disk(self):
|
||||
self.d.update()
|
||||
clover = bdmesg.get_bootloader_uuid()
|
||||
print(clover)
|
||||
self.u.resize(80, 24)
|
||||
self.u.head("Select Default Disk")
|
||||
print(" ")
|
||||
print("1. None")
|
||||
print("2. Boot Disk")
|
||||
if clover:
|
||||
print("3. Booted EFI (Clover/OC)")
|
||||
print(" ")
|
||||
print("M. Main Menu")
|
||||
print("Q. Quit")
|
||||
print(" ")
|
||||
menu = self.u.grab("Please pick a default disk: ")
|
||||
if not len(menu):
|
||||
self.default_disk()
|
||||
menu = menu.lower()
|
||||
if menu in ["1","2"]:
|
||||
self.settings["default_disk"] = [None, "boot"][int(menu)-1]
|
||||
self.flush_settings()
|
||||
return
|
||||
elif menu == "3" and clover:
|
||||
self.settings["default_disk"] = "clover"
|
||||
self.flush_settings()
|
||||
return
|
||||
elif menu == "m":
|
||||
return
|
||||
elif menu == "q":
|
||||
self.u.custom_quit()
|
||||
self.default_disk()
|
||||
|
||||
def get_efi(self):
|
||||
self.d.update()
|
||||
clover = bdmesg.get_bootloader_uuid()
|
||||
i = 0
|
||||
disk_string = ""
|
||||
if not self.full:
|
||||
clover_disk = self.d.get_parent(clover)
|
||||
mounts = self.d.get_mounted_volume_dicts()
|
||||
for d in mounts:
|
||||
i += 1
|
||||
disk_string += "{}. {} ({})".format(i, d["name"], d["identifier"])
|
||||
if self.d.get_parent(d["identifier"]) == clover_disk:
|
||||
# if d["disk_uuid"] == clover:
|
||||
disk_string += " *"
|
||||
disk_string += "\n"
|
||||
else:
|
||||
mounts = self.d.get_disks_and_partitions_dict()
|
||||
disks = mounts.keys()
|
||||
for d in disks:
|
||||
i += 1
|
||||
disk_string+= "{}. {}:\n".format(i, d)
|
||||
parts = mounts[d]["partitions"]
|
||||
part_list = []
|
||||
for p in parts:
|
||||
p_text = " - {} ({})".format(p["name"], p["identifier"])
|
||||
if p["disk_uuid"] == clover:
|
||||
# Got Clover
|
||||
p_text += " *"
|
||||
part_list.append(p_text)
|
||||
if len(part_list):
|
||||
disk_string += "\n".join(part_list) + "\n"
|
||||
height = len(disk_string.split("\n"))+16
|
||||
if height < 24:
|
||||
height = 24
|
||||
self.u.resize(80, height)
|
||||
self.u.head()
|
||||
print(" ")
|
||||
print(disk_string)
|
||||
if not self.full:
|
||||
print("S. Switch to Full Output")
|
||||
else:
|
||||
print("S. Switch to Slim Output")
|
||||
lay = self.settings.get("full_layout", False)
|
||||
l_str = "Slim"
|
||||
if lay:
|
||||
l_str = "Full"
|
||||
print("L. Set As Default Layout (Current: {})".format(l_str))
|
||||
print("B. Mount the Boot Drive's EFI")
|
||||
if clover:
|
||||
print("C. Mount the Booted EFI (Clover/OC)")
|
||||
print("")
|
||||
|
||||
dd = self.settings.get("default_disk", None)
|
||||
if dd == "clover":
|
||||
dd = clover
|
||||
elif dd == "boot":
|
||||
dd = "/"
|
||||
di = self.d.get_identifier(dd)
|
||||
if di:
|
||||
print("D. Pick Default Disk ({} - {})".format(self.d.get_volume_name(di), di))
|
||||
else:
|
||||
print("D. Pick Default Disk (None Set)")
|
||||
|
||||
am = self.settings.get("after_mount", None)
|
||||
if not am:
|
||||
am = "Return to Menu"
|
||||
print("M. After Mounting: "+am)
|
||||
print("Q. Quit")
|
||||
print(" ")
|
||||
print("(* denotes the booted EFI (Clover/OC))")
|
||||
|
||||
menu = self.u.grab("Pick the drive containing your EFI: ")
|
||||
if not len(menu):
|
||||
if not di:
|
||||
return self.get_efi()
|
||||
return self.d.get_efi(di)
|
||||
menu = menu.lower()
|
||||
if menu == "q":
|
||||
self.u.resize(80,24)
|
||||
self.u.custom_quit()
|
||||
elif menu == "s":
|
||||
self.full ^= True
|
||||
return self.get_efi()
|
||||
elif menu == "b":
|
||||
return self.d.get_efi("/")
|
||||
elif menu == "c" and clover:
|
||||
return self.d.get_efi(clover)
|
||||
elif menu == "m":
|
||||
self.after_mount()
|
||||
return
|
||||
elif menu == "d":
|
||||
self.default_disk()
|
||||
return
|
||||
elif menu == "l":
|
||||
self.settings["full_layout"] = self.full
|
||||
self.flush_settings()
|
||||
return
|
||||
try:
|
||||
disk_iden = int(menu)
|
||||
if not (disk_iden > 0 and disk_iden <= len(mounts)):
|
||||
# out of range!
|
||||
self.u.grab("Invalid disk!", timeout=3)
|
||||
return self.get_efi()
|
||||
if type(mounts) is list:
|
||||
# We have the small list
|
||||
disk = mounts[disk_iden-1]["identifier"]
|
||||
else:
|
||||
# We have the dict
|
||||
disk = mounts.keys()[disk_iden-1]
|
||||
except:
|
||||
disk = menu
|
||||
iden = self.d.get_identifier(disk)
|
||||
name = self.d.get_volume_name(disk)
|
||||
if not iden:
|
||||
self.u.grab("Invalid disk!", timeout=3)
|
||||
return self.get_efi()
|
||||
# Valid disk!
|
||||
return self.d.get_efi(iden)
|
||||
|
||||
def main(self):
|
||||
while True:
|
||||
efi = self.get_efi()
|
||||
if not efi:
|
||||
# Got nothing back
|
||||
continue
|
||||
# Mount the EFI partition
|
||||
self.u.head("Mounting {}".format(efi))
|
||||
print(" ")
|
||||
out = self.d.mount_partition(efi)
|
||||
if out[2] == 0:
|
||||
print(out[0])
|
||||
else:
|
||||
print(out[1])
|
||||
# Check our settings
|
||||
am = self.settings.get("after_mount", None)
|
||||
if not am:
|
||||
continue
|
||||
if "reveal" in am.lower():
|
||||
# Reveal
|
||||
mp = self.d.get_mount_point(efi)
|
||||
if mp:
|
||||
self.r.run({"args":["open", mp]})
|
||||
# Hang out for a couple seconds
|
||||
if not self.settings.get("skip_countdown", False):
|
||||
self.u.grab("", timeout=3)
|
||||
if "quit" in am.lower():
|
||||
# Quit
|
||||
self.u.resize(80,24)
|
||||
self.u.custom_quit()
|
||||
|
||||
def quiet_mount(self, disk_list, unmount=False):
|
||||
ret = 0
|
||||
for disk in disk_list:
|
||||
ident = self.d.get_identifier(disk)
|
||||
if not ident:
|
||||
continue
|
||||
efi = self.d.get_efi(ident)
|
||||
if not efi:
|
||||
continue
|
||||
if unmount:
|
||||
out = self.d.unmount_partition(efi)
|
||||
else:
|
||||
out = self.d.mount_partition(efi)
|
||||
if not out[2] == 0:
|
||||
ret = out[2]
|
||||
exit(ret)
|
||||
|
||||
if __name__ == '__main__':
|
||||
# Setup the cli args
|
||||
parser = argparse.ArgumentParser(prog="MountEFI.command", description="MountEFI - an EFI Mounting Utility by CorpNewt")
|
||||
parser.add_argument("-u", "--unmount", help="unmount instead of mount the passed EFIs", action="store_true")
|
||||
parser.add_argument("-p", "--print-efi", help="prints the disk#s# of the EFI attached to the passed var")
|
||||
parser.add_argument("disks",nargs="*")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
m = MountEFI(settings="./Scripts/settings.json")
|
||||
# Gather defaults
|
||||
unmount = False
|
||||
if args.unmount:
|
||||
unmount = True
|
||||
if args.print_efi:
|
||||
print("{}".format(m.d.get_efi(args.print_efi)))
|
||||
# Check for args
|
||||
if len(args.disks):
|
||||
# We got command line args!
|
||||
m.quiet_mount(args.disks, unmount)
|
||||
elif not args.print_efi:
|
||||
m.main()
|
||||
@@ -0,0 +1,18 @@
|
||||
# MountEFI
|
||||
An *even* more robust edition of my previous MountEFI scripts.
|
||||
|
||||
Other scripts can call this script to do a silent mount - and receive a 0 on succes, or 1 (or higher) on failure.
|
||||
|
||||
For example: If another script calls `MountEFI.command disk0` then my script would mount without user interaction, and return a 0 on success, or a 1 on failure. This can also take multiple EFIs to mount - `MountEFI.command disk0 / disk3` would mount the EFIs connected to `disk0`, the boot drive (`/`), and `disk3` if they exist.
|
||||
|
||||
***
|
||||
|
||||
## To install:
|
||||
|
||||
Do the following one line at a time in Terminal:
|
||||
|
||||
git clone https://github.com/corpnewt/MountEFI
|
||||
cd MountEFI
|
||||
chmod +x MountEFI.command
|
||||
|
||||
Then run with either `./MountEFI.command` or by double-clicking *MountEFI.command*
|
||||
@@ -0,0 +1,4 @@
|
||||
from os.path import dirname, basename, isfile
|
||||
import glob
|
||||
modules = glob.glob(dirname(__file__)+"/*.py")
|
||||
__all__ = [ basename(f)[:-3] for f in modules if isfile(f) and not f.endswith('__init__.py')]
|
||||
Binary file not shown.
Executable
+70
@@ -0,0 +1,70 @@
|
||||
import binascii, subprocess, sys
|
||||
|
||||
def get_clover_uuid():
|
||||
bd = bdmesg()
|
||||
if not len(bd):
|
||||
return ""
|
||||
# Get bdmesg output - then parse for SelfDevicePath
|
||||
if not "SelfDevicePath=" in bd:
|
||||
# Not found
|
||||
return ""
|
||||
try:
|
||||
# Split to just the contents of that line
|
||||
line = bd.split("SelfDevicePath=")[1].split("\n")[0]
|
||||
# Get the HD section
|
||||
hd = line.split("HD(")[1].split(")")[0]
|
||||
# Get the UUID
|
||||
uuid = hd.split(",")[2]
|
||||
return uuid
|
||||
except:
|
||||
pass
|
||||
return ""
|
||||
|
||||
def get_oc_uuid():
|
||||
p = subprocess.Popen(["nvram","4D1FDA02-38C7-4A6A-9CC6-4BCCA8B30102:boot-path"], shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
oc, oe = p.communicate()
|
||||
oc = _decode(oc)
|
||||
try:
|
||||
path = oc.split("GPT,")[1].split(",")[0]
|
||||
except:
|
||||
path = ""
|
||||
return path
|
||||
|
||||
def get_bootloader_uuid():
|
||||
b_uuid = get_clover_uuid()
|
||||
if not b_uuid:
|
||||
b_uuid = get_oc_uuid()
|
||||
return b_uuid
|
||||
|
||||
def bdmesg(just_clover = True):
|
||||
b = "" if just_clover else _bdmesg(["ioreg","-l","-p","IOService","-w0"])
|
||||
if b == "":
|
||||
b = _bdmesg(["ioreg","-l","-p","IODeviceTree","-w0"])
|
||||
return b
|
||||
|
||||
def _decode(var):
|
||||
if sys.version_info >= (3,0) and isinstance(var, bytes):
|
||||
var = var.decode("utf-8","ignore")
|
||||
return var
|
||||
|
||||
def _bdmesg(comm):
|
||||
# Runs ioreg -l -p IODeviceTree -w0 and searches for "boot-log"
|
||||
p = subprocess.Popen(comm, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
bd, be = p.communicate()
|
||||
bd = _decode(bd)
|
||||
for line in bd.split("\n"):
|
||||
# We're just looking for the "boot-log" property, then we need to format it
|
||||
if not '"boot-log"' in line:
|
||||
# Skip it!
|
||||
continue
|
||||
# Must have found it - let's try to split it, then get the hex data and process it
|
||||
try:
|
||||
# Split it up, then convert from hex to ascii
|
||||
ascii_bytes = binascii.unhexlify(line.split("<")[1].split(">")[0].encode("utf-8"))
|
||||
ascii_bytes = _decode(ascii_bytes)
|
||||
return ascii_bytes
|
||||
except:
|
||||
# Failed to convert
|
||||
return ""
|
||||
# Didn't find it
|
||||
return ""
|
||||
Binary file not shown.
Executable
+469
@@ -0,0 +1,469 @@
|
||||
import subprocess, plistlib, sys, os, time, json
|
||||
sys.path.append(os.path.abspath(os.path.dirname(os.path.realpath(__file__))))
|
||||
import run
|
||||
if sys.version_info < (3,0):
|
||||
# Force use of StringIO instead of cStringIO as the latter
|
||||
# has issues with Unicode strings
|
||||
from StringIO import StringIO
|
||||
|
||||
class Disk:
|
||||
|
||||
def __init__(self):
|
||||
self.r = run.Run()
|
||||
self.diskutil = self.get_diskutil()
|
||||
self.os_version = ".".join(
|
||||
self.r.run({"args":["sw_vers", "-productVersion"]})[0].split(".")[:2]
|
||||
)
|
||||
self.full_os_version = self.r.run({"args":["sw_vers", "-productVersion"]})[0]
|
||||
if len(self.full_os_version.split(".")) < 3:
|
||||
# Add .0 in case of 10.14
|
||||
self.full_os_version += ".0"
|
||||
self.sudo_mount_version = "10.13.6"
|
||||
self.sudo_mount_types = ["efi"]
|
||||
self.apfs = {}
|
||||
self._update_disks()
|
||||
|
||||
def _get_str(self, val):
|
||||
# Helper method to return a string value based on input type
|
||||
if (sys.version_info < (3,0) and isinstance(val, unicode)) or (sys.version_info >= (3,0) and isinstance(val, bytes)):
|
||||
return val.encode("utf-8")
|
||||
return str(val)
|
||||
|
||||
def _get_plist(self, s):
|
||||
p = {}
|
||||
try:
|
||||
if sys.version_info >= (3, 0):
|
||||
p = plistlib.loads(s.encode("utf-8"))
|
||||
else:
|
||||
# p = plistlib.readPlistFromString(s)
|
||||
# We avoid using readPlistFromString() as that uses
|
||||
# cStringIO and fails when Unicode strings are detected
|
||||
# Don't subclass - keep the parser local
|
||||
from xml.parsers.expat import ParserCreate
|
||||
# Create a new PlistParser object - then we need to set up
|
||||
# the values and parse.
|
||||
pa = plistlib.PlistParser()
|
||||
# We also monkey patch this to encode unicode as utf-8
|
||||
def end_string():
|
||||
d = pa.getData()
|
||||
if isinstance(d,unicode):
|
||||
d = d.encode("utf-8")
|
||||
pa.addObject(d)
|
||||
pa.end_string = end_string
|
||||
parser = ParserCreate()
|
||||
parser.StartElementHandler = pa.handleBeginElement
|
||||
parser.EndElementHandler = pa.handleEndElement
|
||||
parser.CharacterDataHandler = pa.handleData
|
||||
if isinstance(s, unicode):
|
||||
# Encode unicode -> string; use utf-8 for safety
|
||||
s = s.encode("utf-8")
|
||||
# Parse the string
|
||||
parser.Parse(s, 1)
|
||||
p = pa.root
|
||||
except Exception as e:
|
||||
print(e)
|
||||
pass
|
||||
return p
|
||||
|
||||
def _compare_versions(self, vers1, vers2, pad = -1):
|
||||
# Helper method to compare ##.## strings
|
||||
#
|
||||
# vers1 < vers2 = True
|
||||
# vers1 = vers2 = None
|
||||
# vers1 > vers2 = False
|
||||
#
|
||||
# Must be separated with a period
|
||||
|
||||
# Sanitize the pads
|
||||
pad = -1 if not type(pad) is int else pad
|
||||
|
||||
# Cast as strings
|
||||
vers1 = str(vers1)
|
||||
vers2 = str(vers2)
|
||||
|
||||
# Split to lists
|
||||
v1_parts = vers1.split(".")
|
||||
v2_parts = vers2.split(".")
|
||||
|
||||
# Equalize lengths
|
||||
if len(v1_parts) < len(v2_parts):
|
||||
v1_parts.extend([str(pad) for x in range(len(v2_parts) - len(v1_parts))])
|
||||
elif len(v2_parts) < len(v1_parts):
|
||||
v2_parts.extend([str(pad) for x in range(len(v1_parts) - len(v2_parts))])
|
||||
|
||||
# Iterate and compare
|
||||
for i in range(len(v1_parts)):
|
||||
# Remove non-numeric
|
||||
v1 = ''.join(c for c in v1_parts[i] if c.isdigit())
|
||||
v2 = ''.join(c for c in v2_parts[i] if c.isdigit())
|
||||
# If empty - make it a pad var
|
||||
v1 = pad if not len(v1) else v1
|
||||
v2 = pad if not len(v2) else v2
|
||||
# Compare
|
||||
if int(v1) < int(v2):
|
||||
return True
|
||||
elif int(v1) > int(v2):
|
||||
return False
|
||||
# Never differed - return None, must be equal
|
||||
return None
|
||||
|
||||
def update(self):
|
||||
self._update_disks()
|
||||
|
||||
def _update_disks(self):
|
||||
self.disks = self.get_disks()
|
||||
self.disk_text = self.get_disk_text()
|
||||
if self._compare_versions("10.12", self.os_version):
|
||||
self.apfs = self.get_apfs()
|
||||
else:
|
||||
self.apfs = {}
|
||||
|
||||
def get_diskutil(self):
|
||||
# Returns the path to the diskutil binary
|
||||
return self.r.run({"args":["which", "diskutil"]})[0].split("\n")[0].split("\r")[0]
|
||||
|
||||
def get_disks(self):
|
||||
# Returns a dictionary object of connected disks
|
||||
disk_list = self.r.run({"args":[self.diskutil, "list", "-plist"]})[0]
|
||||
return self._get_plist(disk_list)
|
||||
|
||||
def get_disk_text(self):
|
||||
# Returns plain text listing connected disks
|
||||
return self.r.run({"args":[self.diskutil, "list"]})[0]
|
||||
|
||||
def get_disk_info(self, disk):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
disk_list = self.r.run({"args":[self.diskutil, "info", "-plist", disk_id]})[0]
|
||||
return self._get_plist(disk_list)
|
||||
|
||||
def get_disk_fs(self, disk):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
return self.get_disk_info(disk_id).get("FilesystemName", None)
|
||||
|
||||
def get_disk_fs_type(self, disk):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
return self.get_disk_info(disk_id).get("FilesystemType", None)
|
||||
|
||||
def get_apfs(self):
|
||||
# Returns a dictionary object of apfs disks
|
||||
output = self.r.run({"args":"echo y | " + self.diskutil + " apfs list -plist", "shell" : True})
|
||||
if not output[2] == 0:
|
||||
# Error getting apfs info - return an empty dict
|
||||
return {}
|
||||
disk_list = output[0]
|
||||
p_list = disk_list.split("<?xml")
|
||||
if len(p_list) > 1:
|
||||
# We had text before the start - get only the plist info
|
||||
disk_list = "<?xml" + p_list[-1]
|
||||
return self._get_plist(disk_list)
|
||||
|
||||
def is_apfs(self, disk):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
# Takes a disk identifier, and returns whether or not it's apfs
|
||||
for d in self.disks.get("AllDisksAndPartitions", []):
|
||||
if not "APFSVolumes" in d:
|
||||
continue
|
||||
if d.get("DeviceIdentifier", "").lower() == disk_id.lower():
|
||||
return True
|
||||
for a in d.get("APFSVolumes", []):
|
||||
if a.get("DeviceIdentifier", "").lower() == disk_id.lower():
|
||||
return True
|
||||
return False
|
||||
|
||||
def is_apfs_container(self, disk):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
# Takes a disk identifier, and returns whether or not that specific
|
||||
# disk/volume is an APFS Container
|
||||
for d in self.disks.get("AllDisksAndPartitions", []):
|
||||
# Only check partitions
|
||||
for p in d.get("Partitions", []):
|
||||
if disk_id.lower() == p.get("DeviceIdentifier", "").lower():
|
||||
return p.get("Content", "").lower() == "apple_apfs"
|
||||
return False
|
||||
|
||||
def is_cs_container(self, disk):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
# Takes a disk identifier, and returns whether or not that specific
|
||||
# disk/volume is an CoreStorage Container
|
||||
for d in self.disks.get("AllDisksAndPartitions", []):
|
||||
# Only check partitions
|
||||
for p in d.get("Partitions", []):
|
||||
if disk_id.lower() == p.get("DeviceIdentifier", "").lower():
|
||||
return p.get("Content", "").lower() == "apple_corestorage"
|
||||
return False
|
||||
|
||||
def is_core_storage(self, disk):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
if self._get_physical_disk(disk_id, "Logical Volume on "):
|
||||
return True
|
||||
return False
|
||||
|
||||
def get_identifier(self, disk):
|
||||
# Should be able to take a mount point, disk name, or disk identifier,
|
||||
# and return the disk's identifier
|
||||
# Iterate!!
|
||||
if not disk or not len(self._get_str(disk)):
|
||||
return None
|
||||
disk = self._get_str(disk).lower()
|
||||
if disk.startswith("/dev/r"):
|
||||
disk = disk[len("/dev/r"):]
|
||||
elif disk.startswith("/dev/"):
|
||||
disk = disk[len("/dev/"):]
|
||||
if disk in self.disks.get("AllDisks", []):
|
||||
return disk
|
||||
for d in self.disks.get("AllDisksAndPartitions", []):
|
||||
for a in d.get("APFSVolumes", []):
|
||||
if disk in [ self._get_str(a.get(x, "")).lower() for x in ["DeviceIdentifier", "VolumeName", "VolumeUUID", "DiskUUID", "MountPoint"] ]:
|
||||
return a.get("DeviceIdentifier", None)
|
||||
for a in d.get("Partitions", []):
|
||||
if disk in [ self._get_str(a.get(x, "")).lower() for x in ["DeviceIdentifier", "VolumeName", "VolumeUUID", "DiskUUID", "MountPoint"] ]:
|
||||
return a.get("DeviceIdentifier", None)
|
||||
# At this point, we didn't find it
|
||||
return None
|
||||
|
||||
def get_top_identifier(self, disk):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
return disk_id.replace("disk", "didk").split("s")[0].replace("didk", "disk")
|
||||
|
||||
def _get_physical_disk(self, disk, search_term):
|
||||
# Change disk0s1 to disk0
|
||||
our_disk = self.get_top_identifier(disk)
|
||||
our_term = "/dev/" + our_disk
|
||||
found_disk = False
|
||||
our_text = ""
|
||||
for line in self.disk_text.split("\n"):
|
||||
if line.lower().startswith(our_term):
|
||||
found_disk = True
|
||||
continue
|
||||
if not found_disk:
|
||||
continue
|
||||
if line.lower().startswith("/dev/disk"):
|
||||
# At the next disk - bail
|
||||
break
|
||||
if search_term.lower() in line.lower():
|
||||
our_text = line
|
||||
break
|
||||
if not len(our_text):
|
||||
# Nothing found
|
||||
return None
|
||||
our_stores = "".join(our_text.strip().split(search_term)[1:]).split(" ,")
|
||||
if not len(our_stores):
|
||||
return None
|
||||
for store in our_stores:
|
||||
efi = self.get_efi(store)
|
||||
if efi:
|
||||
return store
|
||||
return None
|
||||
|
||||
def get_physical_store(self, disk):
|
||||
# Returns the physical store containing the EFI
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
if not self.is_apfs(disk_id):
|
||||
return None
|
||||
return self._get_physical_disk(disk_id, "Physical Store ")
|
||||
|
||||
def get_core_storage_pv(self, disk):
|
||||
# Returns the core storage physical volume containing the EFI
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
if not self.is_core_storage(disk_id):
|
||||
return None
|
||||
return self._get_physical_disk(disk_id, "Logical Volume on ")
|
||||
|
||||
def get_parent(self, disk):
|
||||
# Disk can be a mount point, disk name, or disk identifier
|
||||
disk_id = self.get_identifier(disk)
|
||||
if self.is_apfs(disk_id):
|
||||
disk_id = self.get_physical_store(disk_id)
|
||||
elif self.is_core_storage(disk_id):
|
||||
disk_id = self.get_core_storage_pv(disk_id)
|
||||
if not disk_id:
|
||||
return None
|
||||
if self.is_apfs(disk_id):
|
||||
# We have apfs - let's get the container ref
|
||||
for a in self.apfs.get("Containers", []):
|
||||
# Check if it's the whole container
|
||||
if a.get("ContainerReference", "").lower() == disk_id.lower():
|
||||
return a["ContainerReference"]
|
||||
# Check through each volume and return the parent's container ref
|
||||
for v in a.get("Volumes", []):
|
||||
if v.get("DeviceIdentifier", "").lower() == disk_id.lower():
|
||||
return a.get("ContainerReference", None)
|
||||
else:
|
||||
# Not apfs - go through all volumes and whole disks
|
||||
for d in self.disks.get("AllDisksAndPartitions", []):
|
||||
if d.get("DeviceIdentifier", "").lower() == disk_id.lower():
|
||||
return d["DeviceIdentifier"]
|
||||
for p in d.get("Partitions", []):
|
||||
if p.get("DeviceIdentifier", "").lower() == disk_id.lower():
|
||||
return d["DeviceIdentifier"]
|
||||
# Didn't find anything
|
||||
return None
|
||||
|
||||
def get_efi(self, disk):
|
||||
disk_id = self.get_parent(self.get_identifier(disk))
|
||||
if not disk_id:
|
||||
return None
|
||||
# At this point - we should have the parent
|
||||
for d in self.disks["AllDisksAndPartitions"]:
|
||||
if d.get("DeviceIdentifier", "").lower() == disk_id.lower():
|
||||
# Found our disk
|
||||
for p in d.get("Partitions", []):
|
||||
if p.get("Content", "").lower() == "efi":
|
||||
return p.get("DeviceIdentifier", None)
|
||||
return None
|
||||
|
||||
def mount_partition(self, disk):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
sudo = False
|
||||
if not self._compare_versions(self.full_os_version, self.sudo_mount_version) and self.get_content(disk_id).lower() in self.sudo_mount_types:
|
||||
sudo = True
|
||||
out = self.r.run({"args":[self.diskutil, "mount", disk_id], "sudo":sudo})
|
||||
self._update_disks()
|
||||
return out
|
||||
|
||||
def unmount_partition(self, disk):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
out = self.r.run({"args":[self.diskutil, "unmount", disk_id]})
|
||||
self._update_disks()
|
||||
return out
|
||||
|
||||
def is_mounted(self, disk):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
m = self.get_mount_point(disk_id)
|
||||
return (m != None and len(m))
|
||||
|
||||
def get_volumes(self):
|
||||
# Returns a list object with all volumes from disks
|
||||
return self.disks.get("VolumesFromDisks", [])
|
||||
|
||||
def _get_value_apfs(self, disk, field, default = None):
|
||||
return self._get_value(disk, field, default, True)
|
||||
|
||||
def _get_value(self, disk, field, default = None, apfs_only = False):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
# Takes a disk identifier, and returns the requested value
|
||||
for d in self.disks.get("AllDisksAndPartitions", []):
|
||||
for a in d.get("APFSVolumes", []):
|
||||
if a.get("DeviceIdentifier", "").lower() == disk_id.lower():
|
||||
return a.get(field, default)
|
||||
if apfs_only:
|
||||
# Skip looking at regular partitions
|
||||
continue
|
||||
if d.get("DeviceIdentifier", "").lower() == disk_id.lower():
|
||||
return d.get(field, default)
|
||||
for a in d.get("Partitions", []):
|
||||
if a.get("DeviceIdentifier", "").lower() == disk_id.lower():
|
||||
return a.get(field, default)
|
||||
return None
|
||||
|
||||
# Getter methods
|
||||
def get_content(self, disk):
|
||||
return self._get_value(disk, "Content")
|
||||
|
||||
def get_volume_name(self, disk):
|
||||
return self._get_value(disk, "VolumeName")
|
||||
|
||||
def get_volume_uuid(self, disk):
|
||||
return self._get_value(disk, "VolumeUUID")
|
||||
|
||||
def get_disk_uuid(self, disk):
|
||||
return self._get_value(disk, "DiskUUID")
|
||||
|
||||
def get_mount_point(self, disk):
|
||||
return self._get_value(disk, "MountPoint")
|
||||
|
||||
def open_mount_point(self, disk, new_window = False):
|
||||
disk_id = self.get_identifier(disk)
|
||||
if not disk_id:
|
||||
return None
|
||||
mount = self.get_mount_point(disk_id)
|
||||
if not mount:
|
||||
return None
|
||||
out = self.r.run({"args":["open", mount]})
|
||||
return out[2] == 0
|
||||
|
||||
def get_mounted_volumes(self):
|
||||
# Returns a list of mounted volumes
|
||||
vol_list = self.r.run({"args":["ls", "-1", "/Volumes"]})[0].split("\n")
|
||||
vol_list = [ x for x in vol_list if x != "" ]
|
||||
return vol_list
|
||||
|
||||
def get_mounted_volume_dicts(self):
|
||||
# Returns a list of dicts of name, identifier, mount point dicts
|
||||
vol_list = []
|
||||
for v in self.get_mounted_volumes():
|
||||
i = self.get_identifier(os.path.join("/Volumes", v))
|
||||
if i == None:
|
||||
i = self.get_identifier("/")
|
||||
if not self.get_volume_name(i) == v:
|
||||
# Not valid and not our boot drive
|
||||
continue
|
||||
vol_list.append({
|
||||
"name" : self.get_volume_name(i),
|
||||
"identifier" : i,
|
||||
"mount_point" : self.get_mount_point(i),
|
||||
"disk_uuid" : self.get_disk_uuid(i),
|
||||
"volume_uuid" : self.get_volume_uuid(i)
|
||||
})
|
||||
return vol_list
|
||||
|
||||
def get_disks_and_partitions_dict(self):
|
||||
# Returns a list of dictionaries like so:
|
||||
# { "disk0" : { "partitions" : [
|
||||
# {
|
||||
# "identifier" : "disk0s1",
|
||||
# "name" : "EFI",
|
||||
# "mount_point" : "/Volumes/EFI"
|
||||
# }
|
||||
# ] } }
|
||||
disks = {}
|
||||
for d in self.disks.get("AllDisks", []):
|
||||
# Get the parent and make sure it has an entry
|
||||
parent = self.get_parent(d)
|
||||
top_disk = self.get_top_identifier(d)
|
||||
if top_disk == d and not self.is_core_storage(d):
|
||||
# Top level, skip
|
||||
continue
|
||||
# Not top level - make sure it's not an apfs container or core storage container
|
||||
if self.is_apfs_container(d):
|
||||
continue
|
||||
if self.is_cs_container(d):
|
||||
continue
|
||||
if not parent in disks:
|
||||
disks[parent] = { "partitions" : [] }
|
||||
disks[parent]["partitions"].append({
|
||||
"name" : self.get_volume_name(d),
|
||||
"identifier" : d,
|
||||
"mount_point" : self.get_mount_point(d),
|
||||
"disk_uuid" : self.get_disk_uuid(d),
|
||||
"volume_uuid" : self.get_volume_uuid(d)
|
||||
})
|
||||
return disks
|
||||
Binary file not shown.
Executable
+130
@@ -0,0 +1,130 @@
|
||||
import sys, os, time, ssl, gzip
|
||||
from io import BytesIO
|
||||
# Python-aware urllib stuff
|
||||
if sys.version_info >= (3, 0):
|
||||
from urllib.request import urlopen, Request
|
||||
else:
|
||||
# Import urllib2 to catch errors
|
||||
import urllib2
|
||||
from urllib2 import urlopen, Request
|
||||
|
||||
class Downloader:
|
||||
|
||||
def __init__(self,**kwargs):
|
||||
self.ua = kwargs.get("useragent",{"User-Agent":"Mozilla"})
|
||||
self.chunk = 1048576 # 1024 x 1024 i.e. 1MiB
|
||||
|
||||
# Provide reasonable default logic to workaround macOS CA file handling
|
||||
cafile = ssl.get_default_verify_paths().openssl_cafile
|
||||
try:
|
||||
# If default OpenSSL CA file does not exist, use that from certifi
|
||||
if not os.path.exists(cafile):
|
||||
import certifi
|
||||
cafile = certifi.where()
|
||||
self.ssl_context = ssl.create_default_context(cafile=cafile)
|
||||
except:
|
||||
# None of the above worked, disable certificate verification for now
|
||||
self.ssl_context = ssl._create_unverified_context()
|
||||
return
|
||||
|
||||
def _decode(self, value, encoding="utf-8", errors="ignore"):
|
||||
# Helper method to only decode if bytes type
|
||||
if sys.version_info >= (3,0) and isinstance(value, bytes):
|
||||
return value.decode(encoding,errors)
|
||||
return value
|
||||
|
||||
def open_url(self, url, headers = None):
|
||||
# Fall back on the default ua if none provided
|
||||
headers = self.ua if headers == None else headers
|
||||
# Wrap up the try/except block so we don't have to do this for each function
|
||||
try:
|
||||
response = urlopen(Request(url, headers=headers), context=self.ssl_context)
|
||||
except Exception as e:
|
||||
# No fixing this - bail
|
||||
return None
|
||||
return response
|
||||
|
||||
def get_size(self, size, suffix=None, use_1024=False, round_to=2, strip_zeroes=False):
|
||||
# size is the number of bytes
|
||||
# suffix is the target suffix to locate (B, KB, MB, etc) - if found
|
||||
# use_2014 denotes whether or not we display in MiB vs MB
|
||||
# round_to is the number of dedimal points to round our result to (0-15)
|
||||
# strip_zeroes denotes whether we strip out zeroes
|
||||
|
||||
# Failsafe in case our size is unknown
|
||||
if size == -1:
|
||||
return "Unknown"
|
||||
# Get our suffixes based on use_1024
|
||||
ext = ["B","KiB","MiB","GiB","TiB","PiB"] if use_1024 else ["B","KB","MB","GB","TB","PB"]
|
||||
div = 1024 if use_1024 else 1000
|
||||
s = float(size)
|
||||
s_dict = {} # Initialize our dict
|
||||
# Iterate the ext list, and divide by 1000 or 1024 each time to setup the dict {ext:val}
|
||||
for e in ext:
|
||||
s_dict[e] = s
|
||||
s /= div
|
||||
# Get our suffix if provided - will be set to None if not found, or if started as None
|
||||
suffix = next((x for x in ext if x.lower() == suffix.lower()),None) if suffix else suffix
|
||||
# Get the largest value that's still over 1
|
||||
biggest = suffix if suffix else next((x for x in ext[::-1] if s_dict[x] >= 1), "B")
|
||||
# Determine our rounding approach - first make sure it's an int; default to 2 on error
|
||||
try:round_to=int(round_to)
|
||||
except:round_to=2
|
||||
round_to = 0 if round_to < 0 else 15 if round_to > 15 else round_to # Ensure it's between 0 and 15
|
||||
bval = round(s_dict[biggest], round_to)
|
||||
# Split our number based on decimal points
|
||||
a,b = str(bval).split(".")
|
||||
# Check if we need to strip or pad zeroes
|
||||
b = b.rstrip("0") if strip_zeroes else b.ljust(round_to,"0") if round_to > 0 else ""
|
||||
return "{:,}{} {}".format(int(a),"" if not b else "."+b,biggest)
|
||||
|
||||
def _progress_hook(self, response, bytes_so_far, total_size):
|
||||
if total_size > 0:
|
||||
percent = float(bytes_so_far) / total_size
|
||||
percent = round(percent*100, 2)
|
||||
t_s = self.get_size(total_size)
|
||||
try: b_s = self.get_size(bytes_so_far, t_s.split(" ")[1])
|
||||
except: b_s = self.get_size(bytes_so_far)
|
||||
sys.stdout.write("Downloaded {} of {} ({:.2f}%)\r".format(b_s, t_s, percent))
|
||||
else:
|
||||
b_s = self.get_size(bytes_so_far)
|
||||
sys.stdout.write("Downloaded {}\r".format(b_s))
|
||||
|
||||
def get_string(self, url, progress = True, headers = None, expand_gzip = True):
|
||||
response = self.get_bytes(url,progress,headers,expand_gzip)
|
||||
if response == None: return None
|
||||
return self._decode(response)
|
||||
|
||||
def get_bytes(self, url, progress = True, headers = None, expand_gzip = True):
|
||||
response = self.open_url(url, headers)
|
||||
if response == None: return None
|
||||
bytes_so_far = 0
|
||||
try: total_size = int(response.headers['Content-Length'])
|
||||
except: total_size = -1
|
||||
chunk_so_far = b""
|
||||
while True:
|
||||
chunk = response.read(self.chunk)
|
||||
bytes_so_far += len(chunk)
|
||||
if progress: self._progress_hook(response, bytes_so_far, total_size)
|
||||
if not chunk: break
|
||||
chunk_so_far += chunk
|
||||
if expand_gzip and response.headers.get("Content-Encoding","unknown").lower() == "gzip":
|
||||
fileobj = BytesIO(chunk_so_far)
|
||||
gfile = gzip.GzipFile(fileobj=fileobj)
|
||||
return gfile.read()
|
||||
return chunk_so_far
|
||||
|
||||
def stream_to_file(self, url, file_path, progress = True, headers = None):
|
||||
response = self.open_url(url, headers)
|
||||
if response == None: return None
|
||||
bytes_so_far = 0
|
||||
try: total_size = int(response.headers['Content-Length'])
|
||||
except: total_size = -1
|
||||
with open(file_path, 'wb') as f:
|
||||
while True:
|
||||
chunk = response.read(self.chunk)
|
||||
bytes_so_far += len(chunk)
|
||||
if progress: self._progress_hook(response, bytes_so_far, total_size)
|
||||
if not chunk: break
|
||||
f.write(chunk)
|
||||
return file_path if os.path.exists(file_path) else None
|
||||
Binary file not shown.
@@ -0,0 +1,39 @@
|
||||
import os, sys
|
||||
sys.path.append(os.path.abspath(os.path.dirname(os.path.realpath(__file__))))
|
||||
import run, utils
|
||||
|
||||
class Rebuild:
|
||||
def __init__(self):
|
||||
self.r = run.Run()
|
||||
return
|
||||
|
||||
def _compare_versions(self, vers1, vers2):
|
||||
# Helper method to compare ##.## strings
|
||||
#
|
||||
# vers1 < vers2 = True
|
||||
# vers1 = vers2 = None
|
||||
# vers1 > vers2 = False
|
||||
#
|
||||
try:
|
||||
v1_parts = vers1.split(".")
|
||||
v2_parts = vers2.split(".")
|
||||
except:
|
||||
# Formatted wrong - return None
|
||||
return None
|
||||
for i in range(len(v1_parts)):
|
||||
if int(v1_parts[i]) < int(v2_parts[i]):
|
||||
return True
|
||||
elif int(v1_parts[i]) > int(v2_parts[i]):
|
||||
return False
|
||||
# Never differed - return None, must be equal
|
||||
return None
|
||||
|
||||
def rebuild(self, stream = True):
|
||||
# Get os version
|
||||
os_vers = self.r.run({"args":["sw_vers", "-productVersion"]})[0]
|
||||
if self._compare_versions(os_vers, "10.11.0") == True:
|
||||
# We're on an OS version prior to 10.11
|
||||
return self.r.run({"args":"sudo touch /System/Library/Extensions && sudo kextcache -u /", "stream" : stream, "shell" : True})
|
||||
else:
|
||||
# 10.11 or above
|
||||
return self.r.run({"args":"sudo kextcache -i / && sudo kextcache -u /", "stream" : stream, "shell" : True})
|
||||
Binary file not shown.
Executable
+70
@@ -0,0 +1,70 @@
|
||||
import sys, os
|
||||
sys.path.append(os.path.abspath(os.path.dirname(os.path.realpath(__file__))))
|
||||
import run
|
||||
|
||||
class Reveal:
|
||||
|
||||
def __init__(self):
|
||||
self.r = run.Run()
|
||||
return
|
||||
|
||||
def get_parent(self, path):
|
||||
return os.path.normpath(os.path.join(path, os.pardir))
|
||||
|
||||
def reveal(self, path, new_window = False):
|
||||
# Reveals the passed path in Finder - only works on macOS
|
||||
if not sys.platform == "darwin":
|
||||
return ("", "macOS Only", 1)
|
||||
if not path:
|
||||
# No path sent - nothing to reveal
|
||||
return ("", "No path specified", 1)
|
||||
# Build our script - then convert it to a single line task
|
||||
if not os.path.exists(path):
|
||||
# Not real - bail
|
||||
return ("", "{} - doesn't exist".format(path), 1)
|
||||
# Get the absolute path
|
||||
path = os.path.abspath(path)
|
||||
command = ["osascript"]
|
||||
if new_window:
|
||||
command.extend([
|
||||
"-e", "set p to \"{}\"".format(path.replace("\"", "\\\"")),
|
||||
"-e", "tell application \"Finder\"",
|
||||
"-e", "reveal POSIX file p as text",
|
||||
"-e", "activate",
|
||||
"-e", "end tell"
|
||||
])
|
||||
else:
|
||||
if path == self.get_parent(path):
|
||||
command.extend([
|
||||
"-e", "set p to \"{}\"".format(path.replace("\"", "\\\"")),
|
||||
"-e", "tell application \"Finder\"",
|
||||
"-e", "reopen",
|
||||
"-e", "activate",
|
||||
"-e", "set target of window 1 to (POSIX file p as text)",
|
||||
"-e", "end tell"
|
||||
])
|
||||
else:
|
||||
command.extend([
|
||||
"-e", "set o to \"{}\"".format(self.get_parent(path).replace("\"", "\\\"")),
|
||||
"-e", "set p to \"{}\"".format(path.replace("\"", "\\\"")),
|
||||
"-e", "tell application \"Finder\"",
|
||||
"-e", "reopen",
|
||||
"-e", "activate",
|
||||
"-e", "set target of window 1 to (POSIX file o as text)",
|
||||
"-e", "select (POSIX file p as text)",
|
||||
"-e", "end tell"
|
||||
])
|
||||
return self.r.run({"args" : command})
|
||||
|
||||
def notify(self, title = None, subtitle = None, sound = None):
|
||||
# Sends a notification
|
||||
if not title:
|
||||
return ("", "Malformed dict", 1)
|
||||
# Build our notification
|
||||
n_text = "display notification with title \"{}\"".format(title.replace("\"", "\\\""))
|
||||
if subtitle:
|
||||
n_text += " subtitle \"{}\"".format(subtitle.replace("\"", "\\\""))
|
||||
if sound:
|
||||
n_text += " sound name \"{}\"".format(sound.replace("\"", "\\\""))
|
||||
command = ["osascript", "-e", n_text]
|
||||
return self.r.run({"args" : command})
|
||||
Binary file not shown.
Executable
+151
@@ -0,0 +1,151 @@
|
||||
import sys, subprocess, time, threading, shlex
|
||||
try:
|
||||
from Queue import Queue, Empty
|
||||
except:
|
||||
from queue import Queue, Empty
|
||||
|
||||
ON_POSIX = 'posix' in sys.builtin_module_names
|
||||
|
||||
class Run:
|
||||
|
||||
def __init__(self):
|
||||
return
|
||||
|
||||
def _read_output(self, pipe, q):
|
||||
try:
|
||||
for line in iter(lambda: pipe.read(1), b''):
|
||||
q.put(line)
|
||||
except ValueError:
|
||||
pass
|
||||
pipe.close()
|
||||
|
||||
def _create_thread(self, output):
|
||||
# Creates a new queue and thread object to watch based on the output pipe sent
|
||||
q = Queue()
|
||||
t = threading.Thread(target=self._read_output, args=(output, q))
|
||||
t.daemon = True
|
||||
return (q,t)
|
||||
|
||||
def _stream_output(self, comm, shell = False):
|
||||
output = error = ""
|
||||
p = None
|
||||
try:
|
||||
if shell and type(comm) is list:
|
||||
comm = " ".join(shlex.quote(x) for x in comm)
|
||||
if not shell and type(comm) is str:
|
||||
comm = shlex.split(comm)
|
||||
p = subprocess.Popen(comm, shell=shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0, universal_newlines=True, close_fds=ON_POSIX)
|
||||
# Setup the stdout thread/queue
|
||||
q,t = self._create_thread(p.stdout)
|
||||
qe,te = self._create_thread(p.stderr)
|
||||
# Start both threads
|
||||
t.start()
|
||||
te.start()
|
||||
|
||||
while True:
|
||||
c = z = ""
|
||||
try: c = q.get_nowait()
|
||||
except Empty: pass
|
||||
else:
|
||||
sys.stdout.write(c)
|
||||
output += c
|
||||
sys.stdout.flush()
|
||||
try: z = qe.get_nowait()
|
||||
except Empty: pass
|
||||
else:
|
||||
sys.stderr.write(z)
|
||||
error += z
|
||||
sys.stderr.flush()
|
||||
if not c==z=="": continue # Keep going until empty
|
||||
# No output - see if still running
|
||||
p.poll()
|
||||
if p.returncode != None:
|
||||
# Subprocess ended
|
||||
break
|
||||
# No output, but subprocess still running - stall for 20ms
|
||||
time.sleep(0.02)
|
||||
|
||||
o, e = p.communicate()
|
||||
return (output+o, error+e, p.returncode)
|
||||
except:
|
||||
if p:
|
||||
try: o, e = p.communicate()
|
||||
except: o = e = ""
|
||||
return (output+o, error+e, p.returncode)
|
||||
return ("", "Command not found!", 1)
|
||||
|
||||
def _decode(self, value, encoding="utf-8", errors="ignore"):
|
||||
# Helper method to only decode if bytes type
|
||||
if sys.version_info >= (3,0) and isinstance(value, bytes):
|
||||
return value.decode(encoding,errors)
|
||||
return value
|
||||
|
||||
def _run_command(self, comm, shell = False):
|
||||
c = None
|
||||
try:
|
||||
if shell and type(comm) is list:
|
||||
comm = " ".join(shlex.quote(x) for x in comm)
|
||||
if not shell and type(comm) is str:
|
||||
comm = shlex.split(comm)
|
||||
p = subprocess.Popen(comm, shell=shell, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
c = p.communicate()
|
||||
except:
|
||||
if c == None:
|
||||
return ("", "Command not found!", 1)
|
||||
return (self._decode(c[0]), self._decode(c[1]), p.returncode)
|
||||
|
||||
def run(self, command_list, leave_on_fail = False):
|
||||
# Command list should be an array of dicts
|
||||
if type(command_list) is dict:
|
||||
# We only have one command
|
||||
command_list = [command_list]
|
||||
output_list = []
|
||||
for comm in command_list:
|
||||
args = comm.get("args", [])
|
||||
shell = comm.get("shell", False)
|
||||
stream = comm.get("stream", False)
|
||||
sudo = comm.get("sudo", False)
|
||||
stdout = comm.get("stdout", False)
|
||||
stderr = comm.get("stderr", False)
|
||||
mess = comm.get("message", None)
|
||||
show = comm.get("show", False)
|
||||
|
||||
if not mess == None:
|
||||
print(mess)
|
||||
|
||||
if not len(args):
|
||||
# nothing to process
|
||||
continue
|
||||
if sudo:
|
||||
# Check if we have sudo
|
||||
out = self._run_command(["which", "sudo"])
|
||||
if "sudo" in out[0]:
|
||||
# Can sudo
|
||||
if type(args) is list:
|
||||
args.insert(0, out[0].replace("\n", "")) # add to start of list
|
||||
elif type(args) is str:
|
||||
args = out[0].replace("\n", "") + " " + args # add to start of string
|
||||
|
||||
if show:
|
||||
print(" ".join(args))
|
||||
|
||||
if stream:
|
||||
# Stream it!
|
||||
out = self._stream_output(args, shell)
|
||||
else:
|
||||
# Just run and gather output
|
||||
out = self._run_command(args, shell)
|
||||
if stdout and len(out[0]):
|
||||
print(out[0])
|
||||
if stderr and len(out[1]):
|
||||
print(out[1])
|
||||
# Append output
|
||||
output_list.append(out)
|
||||
# Check for errors
|
||||
if leave_on_fail and out[2] != 0:
|
||||
# Got an error - leave
|
||||
break
|
||||
if len(output_list) == 1:
|
||||
# We only ran one command - just return that output
|
||||
return output_list[0]
|
||||
return output_list
|
||||
Binary file not shown.
@@ -0,0 +1 @@
|
||||
{"after_mount": "Reveal and Quit", "default_disk": null, "full_layout": false, "skip_countdown": false}
|
||||
Executable
+255
@@ -0,0 +1,255 @@
|
||||
import sys, os, time, re, json, datetime, ctypes, subprocess
|
||||
|
||||
if os.name == "nt":
|
||||
# Windows
|
||||
import msvcrt
|
||||
else:
|
||||
# Not Windows \o/
|
||||
import select
|
||||
|
||||
class Utils:
|
||||
|
||||
def __init__(self, name = "Python Script"):
|
||||
self.name = name
|
||||
# Init our colors before we need to print anything
|
||||
cwd = os.getcwd()
|
||||
os.chdir(os.path.dirname(os.path.realpath(__file__)))
|
||||
if os.path.exists("colors.json"):
|
||||
self.colors_dict = json.load(open("colors.json"))
|
||||
else:
|
||||
self.colors_dict = {}
|
||||
os.chdir(cwd)
|
||||
|
||||
def check_admin(self):
|
||||
# Returns whether or not we're admin
|
||||
try:
|
||||
is_admin = os.getuid() == 0
|
||||
except AttributeError:
|
||||
is_admin = ctypes.windll.shell32.IsUserAnAdmin() != 0
|
||||
return is_admin
|
||||
|
||||
def elevate(self, file):
|
||||
# Runs the passed file as admin
|
||||
if self.check_admin():
|
||||
return
|
||||
if os.name == "nt":
|
||||
ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, file, None, 1)
|
||||
else:
|
||||
try:
|
||||
p = subprocess.Popen(["which", "sudo"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
c = p.communicate()[0].decode("utf-8", "ignore").replace("\n", "")
|
||||
os.execv(c, [ sys.executable, 'python'] + sys.argv)
|
||||
except:
|
||||
exit(1)
|
||||
|
||||
def compare_versions(self, vers1, vers2, **kwargs):
|
||||
# Helper method to compare ##.## strings
|
||||
#
|
||||
# vers1 < vers2 = True
|
||||
# vers1 = vers2 = None
|
||||
# vers1 > vers2 = False
|
||||
|
||||
# Sanitize the pads
|
||||
pad = str(kwargs.get("pad", ""))
|
||||
sep = str(kwargs.get("separator", "."))
|
||||
|
||||
ignore_case = kwargs.get("ignore_case", True)
|
||||
|
||||
# Cast as strings
|
||||
vers1 = str(vers1)
|
||||
vers2 = str(vers2)
|
||||
|
||||
if ignore_case:
|
||||
vers1 = vers1.lower()
|
||||
vers2 = vers2.lower()
|
||||
|
||||
# Split and pad lists
|
||||
v1_parts, v2_parts = self.pad_length(vers1.split(sep), vers2.split(sep))
|
||||
|
||||
# Iterate and compare
|
||||
for i in range(len(v1_parts)):
|
||||
# Remove non-numeric
|
||||
v1 = ''.join(c.lower() for c in v1_parts[i] if c.isalnum())
|
||||
v2 = ''.join(c.lower() for c in v2_parts[i] if c.isalnum())
|
||||
# Equalize the lengths
|
||||
v1, v2 = self.pad_length(v1, v2)
|
||||
# Compare
|
||||
if str(v1) < str(v2):
|
||||
return True
|
||||
elif str(v1) > str(v2):
|
||||
return False
|
||||
# Never differed - return None, must be equal
|
||||
return None
|
||||
|
||||
def pad_length(self, var1, var2, pad = "0"):
|
||||
# Pads the vars on the left side to make them equal length
|
||||
pad = "0" if len(str(pad)) < 1 else str(pad)[0]
|
||||
if not type(var1) == type(var2):
|
||||
# Type mismatch! Just return what we got
|
||||
return (var1, var2)
|
||||
if len(var1) < len(var2):
|
||||
if type(var1) is list:
|
||||
var1.extend([str(pad) for x in range(len(var2) - len(var1))])
|
||||
else:
|
||||
var1 = "{}{}".format((pad*(len(var2)-len(var1))), var1)
|
||||
elif len(var2) < len(var1):
|
||||
if type(var2) is list:
|
||||
var2.extend([str(pad) for x in range(len(var1) - len(var2))])
|
||||
else:
|
||||
var2 = "{}{}".format((pad*(len(var1)-len(var2))), var2)
|
||||
return (var1, var2)
|
||||
|
||||
def check_path(self, path):
|
||||
# Let's loop until we either get a working path, or no changes
|
||||
test_path = path
|
||||
last_path = None
|
||||
while True:
|
||||
# Bail if we've looped at least once and the path didn't change
|
||||
if last_path != None and last_path == test_path: return None
|
||||
last_path = test_path
|
||||
# Check if we stripped everything out
|
||||
if not len(test_path): return None
|
||||
# Check if we have a valid path
|
||||
if os.path.exists(test_path):
|
||||
return os.path.abspath(test_path)
|
||||
# Check for quotes
|
||||
if test_path[0] == test_path[-1] and test_path[0] in ('"',"'"):
|
||||
test_path = test_path[1:-1]
|
||||
continue
|
||||
# Check for a tilde and expand if needed
|
||||
if test_path[0] == "~":
|
||||
tilde_expanded = os.path.expanduser(test_path)
|
||||
if tilde_expanded != test_path:
|
||||
# Got a change
|
||||
test_path = tilde_expanded
|
||||
continue
|
||||
# Let's check for spaces - strip from the left first, then the right
|
||||
if test_path[0] in (" ","\t"):
|
||||
test_path = test_path[1:]
|
||||
continue
|
||||
if test_path[-1] in (" ","\t"):
|
||||
test_path = test_path[:-1]
|
||||
continue
|
||||
# Maybe we have escapes to handle?
|
||||
test_path = "\\".join([x.replace("\\", "") for x in test_path.split("\\\\")])
|
||||
|
||||
def grab(self, prompt, **kwargs):
|
||||
# Takes a prompt, a default, and a timeout and shows it with that timeout
|
||||
# returning the result
|
||||
timeout = kwargs.get("timeout", 0)
|
||||
default = kwargs.get("default", None)
|
||||
# If we don't have a timeout - then skip the timed sections
|
||||
if timeout <= 0:
|
||||
if sys.version_info >= (3, 0):
|
||||
return input(prompt)
|
||||
else:
|
||||
return str(raw_input(prompt))
|
||||
# Write our prompt
|
||||
sys.stdout.write(prompt)
|
||||
sys.stdout.flush()
|
||||
if os.name == "nt":
|
||||
start_time = time.time()
|
||||
i = ''
|
||||
while True:
|
||||
if msvcrt.kbhit():
|
||||
c = msvcrt.getche()
|
||||
if ord(c) == 13: # enter_key
|
||||
break
|
||||
elif ord(c) >= 32: #space_char
|
||||
i += c
|
||||
if len(i) == 0 and (time.time() - start_time) > timeout:
|
||||
break
|
||||
else:
|
||||
i, o, e = select.select( [sys.stdin], [], [], timeout )
|
||||
if i:
|
||||
i = sys.stdin.readline().strip()
|
||||
print('') # needed to move to next line
|
||||
if len(i) > 0:
|
||||
return i
|
||||
else:
|
||||
return default
|
||||
|
||||
def cls(self):
|
||||
os.system('cls' if os.name=='nt' else 'clear')
|
||||
|
||||
def cprint(self, message, **kwargs):
|
||||
strip_colors = kwargs.get("strip_colors", False)
|
||||
if os.name == "nt":
|
||||
strip_colors = True
|
||||
reset = u"\u001b[0m"
|
||||
# Requires sys import
|
||||
for c in self.colors:
|
||||
if strip_colors:
|
||||
message = message.replace(c["find"], "")
|
||||
else:
|
||||
message = message.replace(c["find"], c["replace"])
|
||||
if strip_colors:
|
||||
return message
|
||||
sys.stdout.write(message)
|
||||
print(reset)
|
||||
|
||||
# Needs work to resize the string if color chars exist
|
||||
'''# Header drawing method
|
||||
def head(self, text = None, width = 55):
|
||||
if text == None:
|
||||
text = self.name
|
||||
self.cls()
|
||||
print(" {}".format("#"*width))
|
||||
len_text = self.cprint(text, strip_colors=True)
|
||||
mid_len = int(round(width/2-len(len_text)/2)-2)
|
||||
middle = " #{}{}{}#".format(" "*mid_len, len_text, " "*((width - mid_len - len(len_text))-2))
|
||||
if len(middle) > width+1:
|
||||
# Get the difference
|
||||
di = len(middle) - width
|
||||
# Add the padding for the ...#
|
||||
di += 3
|
||||
# Trim the string
|
||||
middle = middle[:-di]
|
||||
newlen = len(middle)
|
||||
middle += "...#"
|
||||
find_list = [ c["find"] for c in self.colors ]
|
||||
|
||||
# Translate colored string to len
|
||||
middle = middle.replace(len_text, text + self.rt_color) # always reset just in case
|
||||
self.cprint(middle)
|
||||
print("#"*width)'''
|
||||
|
||||
# Header drawing method
|
||||
def head(self, text = None, width = 55):
|
||||
if text == None:
|
||||
text = self.name
|
||||
self.cls()
|
||||
print(" {}".format("#"*width))
|
||||
mid_len = int(round(width/2-len(text)/2)-2)
|
||||
middle = " #{}{}{}#".format(" "*mid_len, text, " "*((width - mid_len - len(text))-2))
|
||||
if len(middle) > width+1:
|
||||
# Get the difference
|
||||
di = len(middle) - width
|
||||
# Add the padding for the ...#
|
||||
di += 3
|
||||
# Trim the string
|
||||
middle = middle[:-di] + "...#"
|
||||
print(middle)
|
||||
print("#"*width)
|
||||
|
||||
def resize(self, width, height):
|
||||
print('\033[8;{};{}t'.format(height, width))
|
||||
|
||||
def custom_quit(self):
|
||||
self.head()
|
||||
print("by CorpNewt\n")
|
||||
print("Thanks for testing it out, for bugs/comments/complaints")
|
||||
print("send me a message on Reddit, or check out my GitHub:\n")
|
||||
print("www.reddit.com/u/corpnewt")
|
||||
print("www.github.com/corpnewt\n")
|
||||
# Get the time and wish them a good morning, afternoon, evening, and night
|
||||
hr = datetime.datetime.now().time().hour
|
||||
if hr > 3 and hr < 12:
|
||||
print("Have a nice morning!\n\n")
|
||||
elif hr >= 12 and hr < 17:
|
||||
print("Have a nice afternoon!\n\n")
|
||||
elif hr >= 17 and hr < 21:
|
||||
print("Have a nice evening!\n\n")
|
||||
else:
|
||||
print("Have a nice night!\n\n")
|
||||
exit(0)
|
||||
Binary file not shown.
+2
-3
@@ -1,14 +1,13 @@
|
||||
# Utilities for T450s
|
||||
# Utilities for T450/T450s
|
||||
|
||||
This is where I'll store useful but optional hacks for the laptop
|
||||
|
||||
## alc_fix
|
||||
This is a fix for Headphone jack not working after sleep. To install this, cd into the repository, then:
|
||||
|
||||
- `sudo spctl --master-disable`
|
||||
- `cd alc_fix`
|
||||
- `chmod +x install.sh`
|
||||
- `./install.sh`
|
||||
|
||||
## DW1820A
|
||||
This folder contains instructions and kexts needed to run DW1820A which enables native WiFi/Bluetooth as well as Airdrop, Instant Hotspot, Hand-off and Continuity.
|
||||
|
||||
|
||||
Binary file not shown.
@@ -1,20 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
|
||||
<plist version="1.0">
|
||||
<dict>
|
||||
<key>KeepAlive</key>
|
||||
<true/>
|
||||
<key>Label</key>
|
||||
<string>com.echo.ALCPlugFix</string>
|
||||
<key>ProgramArguments</key>
|
||||
<array>
|
||||
<string>/usr/bin/ALCPlugFix</string>
|
||||
</array>
|
||||
<key>RunAtLoad</key>
|
||||
<true/>
|
||||
<key>ServiceIPC</key>
|
||||
<false/>
|
||||
<key>WorkingDirectory</key>
|
||||
<string>/usr/bin/</string>
|
||||
</dict>
|
||||
</plist>
|
||||
@@ -0,0 +1,21 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
|
||||
<plist version="1.0">
|
||||
<dict>
|
||||
<key>KeepAlive</key>
|
||||
<true/>
|
||||
<key>Label</key>
|
||||
<string>good.win.ALCPlugFix</string>
|
||||
<key>ProgramArguments</key>
|
||||
<array>
|
||||
<string>/usr/local/bin/ALCPlugFix</string>
|
||||
</array>
|
||||
<key>RunAtLoad</key>
|
||||
<true/>
|
||||
<key>ServiceIPC</key>
|
||||
<false/>
|
||||
<!--Shuold be directory that hda-verb at-->
|
||||
<key>WorkingDirectory</key>
|
||||
<string>/usr/local/bin/</string>
|
||||
</dict>
|
||||
</plist>
|
||||
@@ -1,18 +1,17 @@
|
||||
#!/bin/bash
|
||||
|
||||
DAEMON_PATH=/Library/LaunchDaemons/
|
||||
BIN_PATH=/usr/bin/
|
||||
DAEMON_FILE=com.echo.ALCPlugFix.plist
|
||||
BIN_PATH=/usr/local/bin/
|
||||
DAEMON_FILE=good.win.ALCPlugFix.plist
|
||||
VERB_FILE=hda-verb
|
||||
FIX_FILE=ALCPlugFix
|
||||
|
||||
echo "Installing ALCPlugFix. Root user is required."
|
||||
echo "Installing ALCPlugFix v1.7"
|
||||
|
||||
# check if the root filesystem is writeable (starting with macOS 10.15 Catalina, the root filesystem is read-only by default)
|
||||
if sudo test ! -w "/"; then
|
||||
echo "Root filesystem is not writeable. Remounting as read-write and restarting Finder."
|
||||
sudo mount -uw /
|
||||
sudo killall Finder
|
||||
# check if the directory "usr/local/bin" exist, if not then create the directory
|
||||
|
||||
if [ ! -d "$BIN_PATH" ] ; then
|
||||
mkdir "$BIN_PATH" ;
|
||||
fi
|
||||
|
||||
# stop the daemon if it's already running
|
||||
@@ -22,9 +21,10 @@ if sudo launchctl list | grep --quiet ALCPlugFix; then
|
||||
fi
|
||||
|
||||
# copy over the files to their respective locations (overwrite automatically if files exist)
|
||||
|
||||
sudo cp -f ALCPlugFix $BIN_PATH
|
||||
sudo cp -f hda-verb $BIN_PATH
|
||||
sudo cp -f com.echo.ALCPlugFix.plist $DAEMON_PATH
|
||||
sudo cp -f good.win.ALCPlugFix.plist $DAEMON_PATH
|
||||
|
||||
# set permissions and ownership
|
||||
sudo chmod 755 $BIN_PATH$FIX_FILE
|
||||
|
||||
@@ -2,18 +2,11 @@
|
||||
|
||||
echo "Uninstalling ALCPlugFix. Root user is required."
|
||||
|
||||
# check if the root filesystem is writeable (starting with macOS 10.15 Catalina, the root filesystem is read-only by default)
|
||||
if sudo test ! -w "/"; then
|
||||
echo "Root filesystem is not writeable. Remounting as read-write and restarting Finder."
|
||||
sudo mount -uw /
|
||||
sudo killall Finder
|
||||
fi
|
||||
|
||||
sudo rm /usr/bin/ALCPlugFix
|
||||
sudo rm /usr/bin/hda-verb
|
||||
sudo launchctl unload -w /Library/LaunchDaemons/com.echo.ALCPlugFix.plist
|
||||
sudo launchctl remove com.echo.ALCPlugFix
|
||||
sudo rm /Library/LaunchDaemons/com.echo.ALCPlugFix.plist
|
||||
sudo rm /usr/local/bin/ALCPlugFix
|
||||
sudo rm /usr/local/bin/hda-verb
|
||||
sudo launchctl unload -w /Library/LaunchDaemons/good.win.ALCPlugFix.plist
|
||||
sudo launchctl remove good.win.ALCPlugFix
|
||||
sudo rm /Library/LaunchDaemons/good.win.ALCPlugFix.plist
|
||||
|
||||
echo "Done!"
|
||||
exit 0
|
||||
|
||||
Reference in New Issue
Block a user