# OCCAM - Digital Computational Archive and Curation Service # Copyright (C) 2014-2016 wilkie # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . # This is the init process for x86-64 linux environments running in Docker. import sys import json import subprocess import os import select # Determine the ld.so.cache if not os.path.exists("/etc/ld.so.cache") or os.path.islink("/etc/ld.so.cache"): if os.path.islink("/etc/ld.so.cache"): os.remove("/etc/ld.so.cache") # Create ld.so.cache if os.path.exists("/sbin/ldconfig") or os.path.exists("/usr/bin/ldconfig") or os.path.exists("/bin/ldconfig"): for ldconfig_path in ["/sbin/ldconfig", "/usr/bin/ldconfig", "/bin/ldconfig"]: if os.path.exists(ldconfig_path): with open(os.devnull, 'w') as devnull: p = subprocess.Popen([ldconfig_path, "-v", "/usr/lib", "/usr/lib64", "/usr/lib32", "/lib32", "/lib", "/lib64"], env = {"LD_LIBRARY_PATH": "/usr/lib:/usr/lib64:/usr/lib32:/lib32:/lib:/lib64"}, cwd = "/", stdout = devnull, stderr = devnull) p.communicate() # Establish the user if os.path.exists("/sbin/useradd") or os.path.exists("/usr/bin/useradd") or os.path.exists("/bin/useradd") or os.path.exists("/usr/sbin/useradd"): gid = os.getgid() uid = os.getuid() for groupadd_path in ["/sbin/groupadd", "/usr/bin/groupadd", "/bin/groupadd", "/usr/sbin/groupadd"]: if os.path.exists(groupadd_path): with open(os.devnull, 'w') as devnull: p = subprocess.Popen([groupadd_path, "occam", "--gid", str(gid)], cwd = "/", env = {"LD_LIBRARY_PATH": "/usr/lib:/usr/lib64:/usr/lib32:/lib32:/lib:/lib64"}, stdout = devnull, stderr = devnull) p.communicate() for useradd_path in ["/sbin/useradd", "/usr/bin/useradd", "/bin/useradd", "/usr/sbin/useradd"]: if os.path.exists(useradd_path): with open(os.devnull, 'w') as devnull: p = subprocess.Popen([useradd_path, "occam", "--uid", str(uid), "--gid", str(gid)], cwd = "/", env = {"LD_LIBRARY_PATH": "/usr/lib:/usr/lib64:/usr/lib32:/lib32:/lib:/lib64"}, stdout = devnull, stderr = devnull) p.communicate() # Load object.json (the task manifest) f = open("/home/occam/task/objects/0/task.json") task = json.load(f) f.close() interactive = task.get('interactive', False) processIndex = 0 processCount = len(task.get('running', [])) for process in task.get('running', []): # For each process, determine the environment and run the object processIndex = processIndex + 1 i = 0 for obj in process.get('objects', []): i = i + 1 if 'init' in obj: # Read run command if there is one pass if 'run' in obj and i == len(process.get('objects', [])): # Read the environment variables and run the command env = obj['run'].get('env', {}) command = obj['run'].get('command') if not command is None: if not isinstance(command, list): command = [command] if not command[0].strip().startswith("/"): # Make it refer to the volume of the object if a relative path command[0] = os.path.join(obj.get('paths', {}).get('mount'), command[0].strip()) env["OCCAM_INDEX"] = str(obj.get('index')) env["HOME"] = "/home/occam" if not "PATH" in env: env["PATH"] = "/usr/bin:/bin:/sbin" else: env["PATH"] = "%s:/usr/bin:/bin:/sbin" % env["PATH"] # this is the main running process, wait for it cwd = (obj.get('paths', {}).get('cwd')) if len(sys.argv) > 1 and sys.argv[1] == "console": command = ["/bin/bash"] interactive = True p = None try: if processIndex != processCount: p = subprocess.Popen(command, preexec_fn=os.setsid, env = env, cwd = cwd) p.communicate() elif interactive: import pty import tty import termios import signal import array import fcntl # save original tty setting then set it to raw mode old_tty = termios.tcgetattr(sys.stdin) tty.setraw(sys.stdin.fileno()) # open pseudo-terminal to interact with subprocess host, guest = pty.openpty() def update_size(signum, frame): # Get the terminal size of the real terminal, set it on the pty buf = array.array('h', [0, 0, 0, 0]) fcntl.ioctl(pty.STDOUT_FILENO, termios.TIOCGWINSZ, buf, True) fcntl.ioctl(host, termios.TIOCSWINSZ, buf) if p: p.send_signal(signal.SIGWINCH) signal.signal(signal.SIGWINCH, update_size) update_size(signal.SIGWINCH, None) p = subprocess.Popen(command, preexec_fn=os.setsid, env = env, cwd = cwd, stdin=guest, stdout=guest, stderr=guest, universal_newlines=True) os.close(guest) while p.poll() is None: try: r, w, e = select.select([sys.stdin, host], [], []) if sys.stdin in r: d = os.read(sys.stdin.fileno(), 10240) os.write(host, d) elif host in r: o = None try: o = os.read(host, 10240) except OSError as e: pass if o: os.write(sys.stdout.fileno(), o) except select.error: pass # restore tty settings back termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty) else: stdoutBuffer = "" stderrBuffer = "" stdin = None stdinPath = os.path.join(obj.get('paths', {}).get('localMount'), "stdin") if os.path.exists(stdinPath): stdin = open(stdinPath, "rb") with open(os.path.join(obj.get('paths', {}).get('localMount'), "stdout"), "w+") as sout: with open(os.path.join(obj.get('paths', {}).get('localMount'), "stderr"), "w+") as serr: p = subprocess.Popen(command, env = env, cwd = cwd, stdin = stdin, stdout = subprocess.PIPE, stderr = subprocess.PIPE, bufsize=0) stdout = p.stdout stderr = p.stderr readers = [stdout, stderr] closed = [] while True: readable, writable, exceptional = select.select(readers, [], readers) numRead = 0 for reader in readable: read = reader.read(1) numRead = len(read) if reader is stdout: sout.write(read) stdoutBuffer = stdoutBuffer + read if read == "\n": sys.stdout.write(stdoutBuffer) sys.stdout.flush() stdoutBuffer = "" if reader is stderr: serr.write(read) sout.write(read) stderrBuffer = stderrBuffer + read if read == "\n": sys.stderr.write(stderrBuffer) sys.stderr.flush() stderrBuffer = "" if len(read) == 0: closed.append(reader) readers.remove(reader) if len(readable) == 0 and p.poll() is not None: break if numRead == 0 and len(closed) == 2: break p.communicate() # Augment the run report to add the exit code and success runReportPath = "/home/occam/task/objects/%s/run.json" % (obj.get('index')) runReport = {} if os.path.exists(runReportPath): f = open(runReportPath) runReport = json.load(f) f.close() # Ensure phased reports are stored correctly if isinstance(runReport, list): runReport = { "phases": runReport } runReport['code'] = p.returncode # TODO: detect CTRL+C, etc, and report cancelled status if p.returncode != 0: runReport['status'] = 'failed' else: runReport['status'] = 'finished' # Write back run-report f = open(runReportPath, "w+") f.write(json.dumps(runReport)) f.close() except OSError as e: if e.errno == 2: print >> sys.stderr, "Error %s: No such file or directory: %s" % (str(e.errno), " ".join(command)) else: print >> sys.stderr, "OS Error %s" % (e.errno)