diff options
| -rw-r--r-- | .gitignore | 3 | ||||
| -rw-r--r-- | README.md | 28 | ||||
| -rwxr-xr-x | read_rdp_cert.py | 37 | ||||
| -rw-r--r-- | rrc_lib.py | 205 |
4 files changed, 273 insertions, 0 deletions
diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..24387f2 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +*.pem +__pycache__/ +.*.swp diff --git a/README.md b/README.md new file mode 100644 index 0000000..57349e5 --- /dev/null +++ b/README.md @@ -0,0 +1,28 @@ +# Readme for `read_rdp_cert` + +## Overview +Remote Desktop Protocol (rdp) has the ability to use TLS certificates to encrypt the traffic. Unfortunately, the traditional trick with `openssl s_client -connect rpdhost.example.com:3389` does not work due to how RDP has some communication before engaging the tls components. + +This project can read a packet capture, really any pcap that contains the TLSv1 Certificate protocol, and save from the TLSv1 Certificates packets any pem-format certificates to disk. Of course this project is open-source, so you can adapt it to do whatever you want. + +## Reason for existence +I have not found on the Internet appears how to read RDP certificates. + +## Alternatives +I have not researched these alternatives thoroughly, but they showed possible alternative libraries to use: +* [https://github.com/thy09/isolation/blob/master/load_cert.py](https://github.com/thy09/isolation/blob/master/load_cert.py) +* pyshark lib [https://security.stackexchange.com/questions/123851/how-can-i-extract-the-certificate-from-this-pcap-file](https://security.stackexchange.com/questions/123851/how-can-i-extract-the-certificate-from-this-pcap-file) + +## References +The `iplayer_from_raw` function is almost directly from [cuckoolinux -> network.py](https://github.com/0x71/cuckoo-linux/blob/82263c5df40ebe70dc35976b917293eb54a363af/modules/processing/network.py) and is licensed GPL-3. + +## License +GPL-3 + +## Dependencies +Distro | Packages +--------- | -------------------------------- +Fedora 33 | python3-pyOpenSSL, python3-dpkt +Devuan | python3-openssl, python3-dpkt + +Tested against python 3.9.2 but probably could be lowered if you replace the f"" strings (started with python 3.6). diff --git a/read_rdp_cert.py b/read_rdp_cert.py new file mode 100755 index 0000000..411273a --- /dev/null +++ b/read_rdp_cert.py @@ -0,0 +1,37 @@ +#!/usr/bin/env python3 +# File: read_rdp_cert.py +# Location: https://gitlab.com/bgstack15/read-rdp-cert +# Author: bgstack15 +# Startdate: 2021-07-28 14:02 +# Title: Read RDP Certificate Used from a Packet Capture File +# Purpose: Given pcap input file that contains a TLS HANDSHAKE CERTIFICATE packet, extract out the cert +# History: +# Usage: +# Generate packet capture with: +# sudo tcpdump -w ~/packets.in -n -v -A "port 3389 and (tcp[((tcp[12] & 0xf0) >> 2)] = 0x16)" +# Then visit rdp. +# yes no | xfreerdp destserver.internal.example.com +# Then submit the packet capture file to read_rdp_cert.py +# ./read_rdp_cert.py --pcapfile ~/packets.in +# Reference: +# Improve: +# Add debug level + +# Note: if I need libpath logic, check logout-manager-cli +from rrc_lib import * +import argparse + +read_rdp_cert_version="2021-07-29" + +parser = argparse.ArgumentParser(description="read pcap files and extract TLSv1 Certificate certificates") +parser.add_argument("-p","--pcapfile", required=True, help="Input file. Required.") +parser.add_argument("-V","--version", action="version", version="%(prog)s " + read_rdp_cert_version) + +args = parser.parse_args() + +array = read_pcap_file(args.pcapfile) +for i in array: + save_cert( + data = i, + directory = os.path.curdir + ) diff --git a/rrc_lib.py b/rrc_lib.py new file mode 100644 index 0000000..82ecb27 --- /dev/null +++ b/rrc_lib.py @@ -0,0 +1,205 @@ +#!/usr/bin/env python3 +# File: rrc_lib.py +# Location: https://gitlab.com/bgstack15/read-rdp-cert +# Author: bgstack15 +# Startdate: 2021-07-28 14:02 +# Title: Library for Reading RDP Certificates +# Purpose: Functions for reading packet captures of RDP negotiations to observe what cert the server is using. +# History: +# Usage: +# See read_rdp_cert.py +# References: +# packet capture filter with https://stackoverflow.com/questions/39624745/capture-only-ssl-handshake-with-tcpdump +# https://github.com/0x71/cuckoo-linux/blob/82263c5df40ebe70dc35976b917293eb54a363af/modules/processing/network.py +# Alternatives: +# uses different library https://github.com/thy09/isolation/blob/master/load_cert.py +# another library, not packaged by my distro https://security.stackexchange.com/questions/123851/how-can-i-extract-the-certificate-from-this-pcap-file +# wireshark, with above capture filter and display filter "tls.handshake.type == 11" +# Improve: +# Ensure that this handles a cert chain correctly? +# Operate on streams? +# Wild idea: initiate the network traffic that triggers the RDP and TLS certificate directly? +# Dependencies: +# python3-dpkt, python3-openssl +# python 3.9.2 +import dpkt, os, ssl +from OpenSSL.crypto import load_certificate, FILETYPE_PEM + +def read_pcap_file(infile): + inf = open(infile,"rb") + reader = dpkt.pcap.Reader(inf) + array = [] + count = 0 + help_shown = False + for ts, buf in reader: + count += 1; + print("-- Packet number:",count) + data = None + #array.append({"ts": ts, "buf": buf}) + tcp = dpkt.tcp.TCP() + ip = iplayer_from_raw(buf,reader.datalink()) + if ip.p == dpkt.ip.IP_PROTO_TCP: + tcp = ip.data + if not tcp.data: + print("Not tcp data.") + # THE USEFUL STUFF IS tcp.data + #print("tcp data") + continue + data = tcp.data + # now we can process data + if data: + a = extract_cert2(data) + if a != -1: + for i in a: + array.append(i) + return array + +#class SamplePacket(dpkt.dpkt.Packet): +# __hdr__ = () + +def old_attempt1(): + # try: + # record = dpkt.ssl.TLSRecord(data) + # except dpkt.NeedData: + # print(f"Need data, packet {count}") + # continue + # # this all works if we have already limited the capture filter + # record = dpkt.ssl.RECORD_TYPES[record.type](record.data) + # if not isinstance(record, dpkt.ssl.TLSHandshake): + # print(f"Packet {count} is not tls handshake") + # if isinstance(record, dpkt.ssl.TLSCertificate): + # print(f"Packet {count} is tls certificate") + # try: + # record = dpkt.ssl.TLSCertificate(data) + # except: + # print(f"Packet {count} could not be converted to tlscertificate") + # #a = dpkt.ssl.TLSCertificate() #.unpack(buf=record) + # #a.unpack(buf=record) + # #print(a) + # #if not help_shown: + # # help_shown = True + # # help(record.data) + # packet = SamplePacket(buf=buf) + # #extract_cert(packet) + return 0 + +# iplayer_from_raw ripped from https://github.com/0x71/cuckoo-linux/blob/82263c5df40ebe70dc35976b917293eb54a363af/modules/processing/network.py +def iplayer_from_raw(raw, linktype=1): + """Converts a raw packet to a dpkt packet regarding of link type. + @param raw: raw packet + @param linktype: integer describing link type as expected by dpkt + """ + if linktype == 1: # ethernet + pkt = dpkt.ethernet.Ethernet(raw) + ip = pkt.data + elif linktype == 101: # raw + ip = dpkt.ip.IP(raw) + else: + return -1 + return ip + +def extract_cert2(data): + array = [] + #print(f"Working with {data}") + # hard-coded validation against TLSv1 record because library dpkt sucks. + content_type = data[0] + tls_version = conv(data[1:3]) + length = conv(data[3:5]) + print(f"content_type {content_type}") + print(f"tls_version {tls_version}") + print(f"length {length}") + # so the next [length] contents need to be interpreted + if content_type != 22: + print("This is not a tlsv1 handshake") + return -1 + final_byte = 5+1+length + subdata = data[5:final_byte] + tls_object_count = 0 + print(f"length of subdata: {len(subdata)}") + current_byte = 0 + protocol = 0 + p_length = 0 + MAX_LOOPS = 15 + loop_count = 0 + while(current_byte < len(subdata)): + loop_count += 1 + if loop_count >= MAX_LOOPS: + print("Safety valve; too many loops through TLSv1 handshake inspection.") + break + protocol = subdata[current_byte] + p_length = conv(subdata[current_byte+1:current_byte+4]) + if protocol == 1: + print(f"Client hello, length {p_length}") + elif protocol == 2: + print(f"Server hello, length {p_length}") + elif protocol == 11: + print(f"Certificate, the holy grail, length {p_length}") + # there is a separate field here for certs length, which should be 3 bytes shorter than p_length. + certs_length = conv(subdata[current_byte+4:current_byte+4+3]) + print(f"For all certs here, length is {certs_length}") + # this will need to be written probably to loop when there is a chain of certificates + cert_length = conv(subdata[current_byte+4+3:current_byte+4+3+3]) + print(f"Cert length is {cert_length}") + cert_body = subdata[current_byte+4+3+3:current_byte+4+3+3+cert_length] + array.append(cert_body) + print(f"This cert length is {len(cert_body)}") + #with open("/home/bgstack15/cert.out","wb") as o: + # o.write(cert_body) + elif protocol == 14: + print(f"Server hello done, length {p_length}") + elif protocol == 16: + print(f"Client key exchange, length {p_length}") + elif protocol == 20: + print(f"Change cipher spec, length {p_length}") + else: + print(f"Unknown protocol {protocol}, length {p_length}") + + current_byte += (4 + p_length) + return array + +def conv(hexinput): + return int.from_bytes(hexinput,'big') + +def extract_cert(packet): + #if not isinstance(packet,dpkt.dpkg.Packet) + print("packet['buf'] is",packet['buf']) + try: + record = dpkt.ssl.TLSRecord(packet['buf']) + except dpkt.NeedData: + print("need data") + return -1 + try: + record = dpkt.ssl.TLSRecord(packet['buf']) + print("Found it with packet['buf']") + except: + try: + record = dpkt.ssl.TLSRecord(packet) # maybe just the bytes? + print("Found it with packet") + except: + print("Error: extract_cert needs the packet passed to it.") + return -1 + # so now we have a record object + if isinstance(record, dpkt.ssl.TLSHandshake): + print("found a handshake!") + +def save_cert(data, directory = os.path.curdir): + # the other functions tend to output DER certs + cert = data + # so let's try to convert to PEM + try: + cert = ssl.DER_cert_to_PEM_cert(data) + except: + # but don't try very hard + pass + certificate = load_certificate(FILETYPE_PEM, cert) + subject = certificate.get_subject() + # We only need a simple subject name for my use case. + # from https://stackoverflow.com/questions/57877935/how-to-convert-an-x509name-object-to-a-string-in-python/57878347#57878347 + #subject = "".join("/{0:s}={1:s}".format(name.decode(), value.decode()) for name, value in subject.get_components()) + _, subject = subject.get_components()[0] + subject = subject.decode('utf-8') + print(subject) + outfile = os.path.join(directory,subject+".pem") + print(outfile) + with open(outfile,"w") as o: + o.write(cert) |