Country Finder - NRF24CTF
kurtnettleProblem Statement
To be a good cybersecurity engineer you have to learn writing script in python.
Prerequisites
Python (basic)
socket module (i/o ops enough)
Solution
After connecting to the netcat instance, we will be given a list of coordinates. A very straightforward problem where we just need to perform reverse geocoding. So, we now need to
Parse the output to get the latitude and longitude.
Convert the coordinates to a country name (a.k.a reverse geocode).
Talking with netcat
import socket
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
addr = "chall.cbctf.xyz 34xxx"
server_ip, server_port = addr.split()
client.connect((server_ip, int(server_port)))
Before we start communicating with netcat, we need to create a socket instance and connect to the server using the provided address.
Recieving Message
We have a recv() function that receives data from the socket.
data = client.recv(2048) # buffer size = 2048 bytes
# also when you get a empty data it indicates
# that the client got disconnected (we will need this later :D)
# as we recieved bytes, we need to decode it
msg = data.decode()
We successfully read the output programmatically! 🥳
Hold on, We noticed that we will be given multiple rounds. How do we handle this?!
while True:
try:
data = client.recv(2048)
msg = data.decode("utf-8")
if len(msg) == 0:
break
except ConnectionResetError:
break
We wrapped it in a while loop with a try-except block because when the connection is reset, it throws a ConnectionResetError exception.
Parsing Output
Now we need to parse the output. It's very simple since there is no complex formatting.
def parse_output(msg):
for i in msg.split("\n"):
x = i.split(",")
if len(x) != 2:
continue
lat = x[0].split(":")[1].strip()
lon = x[1].split(":")[1].strip()
We split the lines by
\nIn each line, the latitude and longitude are separated by a comma, so we split it by
,The values of latitude and longitude are separated by a colon, so we split them again by
,
Reverse Geocode
This service is offered by many providers. I initially chose GeoNames.
(What does this have to do with the write-up? Well, I learned about GeoNames from there)
However, it returned 404 errors for various coordinates, so I searched online and found LocationIQ and OpenCage
from requests import get
def fallback_0(lat, lon):
api = (
f"http://api.geonames.org/countryCodeJSON?lat={lat}&lng={lon}&username=[USERNAME]"
)
try:
resp = get(api)
data = resp.json()
return data["countryName"]
except:
return False
def fallback_1(lat, lon):
api = f"https://api.opencagedata.com/geocode/v1/json?q={lat}%2C{lon}&key=[API_KEY]"
try:
resp = get(api)
data = resp.json()
return data["results"][0]["components"]["country"]
except:
return False
def fallback_2(lat, lon):
api = f"https://us1.locationiq.com/v1/reverse?lat={lat}&lon={lon}&format=json&zoom=3&accept-language=en&key=[API_KEY]"
try:
resp = get(api)
data = resp.json()
return data["address"]["country"]
except:
return False
I added multiple fallback APIs because I was getting 404 earlier (just to be safe).
Combining the API with the parsed output
def latlon_2_cn(lat, lon):
cn = fallback_1(lat, lon)
if not cn:
print(f"falling back to 2 - {lat},{lon}")
cn = fallback_2(lat, lon)
if not cn:
print(f"failed to get country name - {lat},{lon}")
input("enter the country name manually:")
return cn
def handle_output(msg):
cns = []
for i in msg.split("\n"):
x = i.split(",")
if len(x) != 2:
continue
lat = x[0].split(":")[1].strip()
lon = x[1].split(":")[1].strip()
cn = latlon_2_cn(lat, lon)
if cn:
cns.append(cn)
else:
print(f"{lat},{lon} - country not found")
cn = input("Enter the country name: ")
if not cn:
break
cns = sorted(cns)
return ",".join(cns)
We have multiple rounds, typically 21-28, and it would be terrible if your script fails to get a country name in the middle of it. So, I added an input() as a last resort.
Sending message
Just as we have a recv() function to receive data from the socket, we also have a send() function to send messages to the socket. Remember, we can only send bytes, so we need to encode() the message first.
msg = "my msg"
data = msg.encode()
client.send(data)
Alright, we have covered all the parts, so let's put all the pieces together.
Source Code
apis.py
from requests import get
def fallback_0(lat, lon):
api = (
f"http://api.geonames.org/countryCodeJSON?lat={lat}&lng={lon}&username=[USERNAME]"
)
try:
resp = get(api)
data = resp.json()
return data["countryName"]
except:
return False
def fallback_1(lat, lon):
api = f"https://api.opencagedata.com/geocode/v1/json?q={lat}%2C{lon}&key=[API_KEY]"
try:
resp = get(api)
data = resp.json()
return data["results"][0]["components"]["country"]
except:
return False
def fallback_2(lat, lon):
api = f"https://us1.locationiq.com/v1/reverse?lat={lat}&lon={lon}&format=json&zoom=3&accept-language=en&key=[API_KEY]"
try:
resp = get(api)
data = resp.json()
return data["address"]["country"]
except:
return False
db.py
import sqlite3
class LocationDB:
def __init__(self):
self.conn = sqlite3.connect("locations.db")
self.cursor = self.conn.cursor()
self.cursor.execute(
"""
CREATE TABLE IF NOT EXISTS locations (
lat TEXT,
lon TEXT,
cn TEXT)
"""
)
def put(self, lat, lon, cn):
self.cursor.execute(
"""
INSERT INTO locations (lat, lon, cn)
VALUES (?, ?, ?)
""",
(lat, lon, cn)
)
self.conn.commit()
def get(self, lat, lon):
self.cursor.execute(
"""
SELECT * FROM locations
WHERE lat = ? AND lon = ?
""",
(lat, lon)
)
res = self.cursor.fetchone()
if res:
return res[2]
main.py
import socket
from apis import *
from db import LocationDB
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
db = LocationDB()
addr = "chall.cbctf.xyz xxxxx"
server_ip, server_port = addr.split()
client.connect((server_ip, int(server_port)))
def latlon_2_cn(lat, lon):
cn = db.get(lat, lon)
if cn:
print(f"found in cache - {lat},{lon},{cn}")
# if not cn:
# print("falling back to 0")
# cn = fallback_0(lat,lon)
if not cn:
print(f"falling back to 1 - {lat},{lon}")
cn = fallback_1(lat, lon)
if not cn:
print(f"falling back to 2 - {lat},{lon}")
cn = fallback_2(lat, lon)
if not cn:
print(f"failed to get country name - {lat},{lon}")
if cn:
db.put(lat, lon, cn)
return cn
def handle_output(msg):
cns = []
for i in msg.split("\n"):
x = i.split(",")
if len(x) != 2:
continue
lat = x[0].split(":")[1].strip()
lon = x[1].split(":")[1].strip()
cn = latlon_2_cn(lat, lon)
if cn:
cns.append(cn)
else:
print(f"{lat},{lon} - country not found")
cn = input("Enter the country name: ")
if not cn:
break
cns = sorted(cns)
return ",".join(cns)
if __name__ == "__main__":
while True:
try:
msg = client.recv(2048).decode("utf-8")
print(msg)
if len(msg) == 0:
break
except ConnectionResetError:
break
finally:
client.close()
msg = handle_output(msg)
print(msg)
data = f"{msg}\n".encode("utf-8")
client.send(data)
Conclusion
I tried sharing all of my thought process because I love to read about how other people think about problems more than just seeing the straightforward solution.
Thanks to my coding skills, I had to add a cache system because the API service allows only a limited number of requests (2.5k/day). If I keep debugging my issues on production, I would need to create a dozen of accounts :)
Databases might sound daunting, but the project I am currently working on had a lot of database work, so it was no big issue for me to integrate it.
During the contest, I couldn't submit my script because I forgot about socket programming and was googling it. My teammates were quick and submitted their script while I was still debugging my script.
Credits
Hashnode - for the amazing platform
GeoNames, OpenCage, LocationIQ - for their free-tier API
EWU - for delaying this write-up (they restricted challenge access after <36 hours)
Subscribe to my newsletter
Read articles from kurtnettle directly inside your inbox. Subscribe to the newsletter, and don't miss out.
Written by