intelligence_platform_guard.py
Anonymous
Python
11/08/2026 12:58 PM
12.4 KB
18
Indexable
05/07/2027 11:58 AM
#!/usr/bin/env python3
"""Reversibly obstruct one macOS IntelligencePlatform storage path.
Requires Python 3.8+ and a filesystem supporting renamex_np(RENAME_SWAP).
Run with sudo from the account whose store is to be changed:
sudo python3 intelligence_platform_guard.py block
sudo python3 intelligence_platform_guard.py status
sudo python3 intelligence_platform_guard.py undo
This is an unsupported workaround, not a global privacy setting. Restart
macOS after block/undo: existing processes may retain open database handles.
No databases are deleted. The preserved directory still contains personal data.
No daemons are killed, no launch settings are changed, and SIP is not modified.
"""
import argparse
import ctypes
import fcntl
import json
import os
from pathlib import Path
import pwd
import signal
import stat
import sys
class Refusal(RuntimeError):
pass
class MacFS:
owner = 0
group = 0
def __init__(self):
self.lib = ctypes.CDLL('/usr/lib/libSystem.B.dylib', use_errno=True)
self.lib.renamex_np.argtypes = [ctypes.c_char_p, ctypes.c_char_p,
ctypes.c_uint]
self.lib.renamex_np.restype = ctypes.c_int
self.lib.fchflags.argtypes = [ctypes.c_int, ctypes.c_uint]
self.lib.fchflags.restype = ctypes.c_int
def rename(self, source, destination, swap):
# Darwin: RENAME_SWAP = 0x2; RENAME_EXCL = 0x4. No unsafe fallback.
flag = 0x2 if swap else 0x4
if self.lib.renamex_np(os.fsencode(source), os.fsencode(destination), flag):
err = ctypes.get_errno()
raise OSError(err, os.strerror(err), str(destination))
def immutable(self, fd, enabled):
flags = os.fstat(fd).st_flags
flags = flags | stat.UF_IMMUTABLE if enabled else flags & ~stat.UF_IMMUTABLE
if self.lib.fchflags(fd, flags):
err = ctypes.get_errno()
raise OSError(err, os.strerror(err))
def is_immutable(self, fd):
return bool(os.fstat(fd).st_flags & stat.UF_IMMUTABLE)
def inspect(path):
try:
return path.lstat() # Never follow a final symlink.
except FileNotFoundError:
return None
def identity(info):
return [info.st_dev, info.st_ino]
def matches(path, expected, kind):
info = inspect(path)
return info is not None and identity(info) == expected and kind(info.st_mode)
def private_file(path, owner):
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
try:
info = os.fstat(fd)
if (not stat.S_ISREG(info.st_mode) or info.st_uid != owner
or info.st_nlink != 1 or info.st_mode & 0o077):
raise Refusal('Unexpected permissions or file type: ' + str(path))
return fd
except BaseException:
os.close(fd)
raise
class Guard:
def __init__(self, library, user_id, fs):
self.library = library
self.user_id = user_id
self.fs = fs
self.target = library / 'IntelligencePlatform'
self.backup = library / 'IntelligencePlatform.block-backup'
self.saved = self.backup / 'saved-entry'
self.manifest = self.backup / 'state.json'
def check_backup(self):
info = self.backup.lstat()
if (not stat.S_ISDIR(info.st_mode) or info.st_uid != self.fs.owner
or stat.S_IMODE(info.st_mode) != 0o700):
raise Refusal('Unexpected backup directory; refusing to use it.')
def load(self):
self.check_backup()
with os.fdopen(private_file(self.manifest, self.fs.owner)) as stream:
data = json.load(stream)
if (data.get('version') != 1 or data.get('user_id') != self.user_id
or data.get('target') != str(self.target)
or any(not isinstance(data.get(k), list) or len(data[k]) != 2
or any(type(v) is not int for v in data[k])
for k in ('original', 'placeholder'))):
raise Refusal('Invalid backup metadata; nothing was overwritten.')
return data
def placeholder_fd(self, path, data):
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
info = os.fstat(fd)
if (not stat.S_ISREG(info.st_mode) or info.st_uid != self.fs.owner
or info.st_gid != self.fs.group
or info.st_size != 0 or info.st_nlink != 1
or stat.S_IMODE(info.st_mode) != 0o444
or identity(info) != data['placeholder']):
os.close(fd)
raise Refusal('The expected empty blocking file has changed.')
return fd
def cleanup(self, data):
# Only the script's empty placeholder and metadata are removed.
# No recursive deletion, and never remove the original directory.
if inspect(self.saved) is not None:
fd = self.placeholder_fd(self.saved, data)
try:
self.fs.immutable(fd, False)
finally:
os.close(fd)
self.saved.unlink()
self.manifest.unlink()
self.backup.rmdir()
def undo(self):
data = self.load()
if matches(self.target, data['original'], stat.S_ISDIR):
# Also recovers an interrupted setup before the exchange.
self.cleanup(data)
return
if not matches(self.saved, data['original'], stat.S_ISDIR):
raise Refusal('The preserved directory has changed; refusing to restore.')
if inspect(self.target) is None:
# Restore only if the name is still unused at the actual syscall.
self.fs.rename(self.saved, self.target, swap=False)
else:
fd = self.placeholder_fd(self.target, data)
try:
self.fs.immutable(fd, False)
try:
self.fs.rename(self.saved, self.target, swap=True)
except BaseException:
self.fs.immutable(fd, True)
raise
finally:
os.close(fd)
if not matches(self.target, data['original'], stat.S_ISDIR):
raise Refusal('Restore verification failed; preserved entries were retained.')
self.cleanup(data)
def block(self):
if inspect(self.backup) is not None:
raise Refusal('Backup already exists. Run status or undo first.')
original = inspect(self.target)
if (original is None or not stat.S_ISDIR(original.st_mode)
or original.st_uid != self.user_id
or original.st_dev != self.library.stat().st_dev
or getattr(original, 'st_flags', 0)
& (stat.UF_IMMUTABLE | stat.SF_IMMUTABLE)):
raise Refusal('Expected an unlocked, user-owned directory at ' + str(self.target))
self.backup.mkdir(mode=0o700)
fd = None
ready = False
try:
fd = os.open(self.saved, os.O_CREAT | os.O_EXCL | os.O_RDWR
| os.O_NOFOLLOW, 0o600)
os.fchown(fd, self.fs.owner, self.fs.group)
os.fchmod(fd, 0o444)
os.fsync(fd)
data = dict(version=1, user_id=self.user_id, target=str(self.target),
original=identity(original), placeholder=identity(os.fstat(fd)))
with self.manifest.open('x') as stream:
json.dump(data, stream)
stream.flush()
os.fsync(stream.fileno())
ready = True
if not matches(self.target, data['original'], stat.S_ISDIR):
raise Refusal('The original directory changed during setup.')
# Atomically exchanges the directory with the root-owned file.
self.fs.rename(self.target, self.saved, swap=True)
# Apply uchg to the open file itself, never to an unchecked pathname.
self.fs.immutable(fd, True)
if (not matches(self.target, data['placeholder'], stat.S_ISREG)
or not matches(self.saved, data['original'], stat.S_ISDIR)
or not self.fs.is_immutable(fd)):
raise Refusal('Blocking verification failed.')
except BaseException:
if ready:
try:
self.undo()
print('Setup failed; the original directory is restored.', file=sys.stderr)
except BaseException as exc:
print('Automatic recovery stopped: ' + str(exc), file=sys.stderr)
print('Preserved entries: ' + str(self.backup), file=sys.stderr)
else:
print('Setup stopped before the exchange. Original data was not moved.',
file=sys.stderr)
print('Staging directory retained: ' + str(self.backup), file=sys.stderr)
raise
finally:
if fd is not None:
os.close(fd)
def status(self):
if inspect(self.backup) is None:
info = inspect(self.target)
if info is not None and stat.S_ISDIR(info.st_mode):
print('Not blocked by this script: the target is a directory.')
else:
print('No managed backup. Target is missing or is not a directory.')
return
data = self.load()
if matches(self.target, data['original'], stat.S_ISDIR):
print('Original directory is active. Run undo to finish staging cleanup.')
elif matches(self.saved, data['original'], stat.S_ISDIR):
fd = self.placeholder_fd(self.target, data)
try:
print('Storage path blocked with root-owned uchg file.'
if self.fs.is_immutable(fd) else
'Placeholder exists but is not immutable. Run undo.')
finally:
os.close(fd)
print('Original directory preserved at: ' + str(self.saved))
else:
raise Refusal('Unexpected state. Preserved entries were left untouched.')
def interrupted(signum, frame):
raise InterruptedError('Interrupted by signal ' + str(signum))
def main():
parser = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument('action', choices=('block', 'undo', 'status'), default='status', nargs='?')
args = parser.parse_args()
if sys.platform != 'darwin':
raise Refusal('macOS only. Nothing was changed.')
if os.geteuid() != 0 or not os.environ.get('SUDO_UID', '').isdigit():
raise Refusal('Run with sudo from your own account, not a root login shell.')
user_id = int(os.environ['SUDO_UID'])
if user_id == 0:
raise Refusal('Refusing to target the root account.')
account = pwd.getpwuid(user_id)
library = Path(account.pw_dir) / 'Library'
info = library.lstat()
if (library != library.resolve() or not stat.S_ISDIR(info.st_mode)
or info.st_uid != user_id or info.st_mode & 0o022):
raise Refusal('Unexpected Library path, ownership, or permissions.')
os.umask(0o077)
# Serializes invocations. This empty lock file remains after undo.
lock = library / '.IntelligencePlatform-guard.lock'
lock_fd = os.open(lock, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600)
try:
info = os.fstat(lock_fd)
if (not stat.S_ISREG(info.st_mode) or info.st_uid != 0
or info.st_nlink != 1 or stat.S_IMODE(info.st_mode) != 0o600
or info.st_size != 0):
raise Refusal('Unexpected lock file; refusing to use it.')
fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
signal.signal(signal.SIGTERM, interrupted)
guard = Guard(library, user_id, MacFS())
getattr(guard, args.action)()
if args.action == 'block':
print('Blocked storage path: ' + str(guard.target))
print('Original directory preserved at: ' + str(guard.saved))
elif args.action == 'undo':
print('Restored original directory: ' + str(guard.target))
if args.action != 'status':
print('Restart macOS now to close any old database handles.')
print('This verifies this storage path only, not all personal-data indexing.')
finally:
os.close(lock_fd)
if __name__ == '__main__':
try:
main()
except (Exception, KeyboardInterrupt) as exc:
print('STOPPED: ' + (str(exc) or 'interrupted'), file=sys.stderr)
print('Do not delete a retained backup or disable SIP to force this workaround.',
file=sys.stderr)
sys.exit(1)
Editor is loading...
Leave a Comment