Add feature to define frequencies to avoid

README.md

@@ -35,6 +35,20 @@ dev_id,osmosdr_device_string
 
 **TX Antenna Ports:** challengectl will transmit by default from the default transmit port on your SDR device, the only port if your SDR only has one transmit port, TX1 on bladeRF 2.0 Micro, TX/RX on a USRP B200 - whichever port is selected by an empty string passed to the antenna parameter for the Osmocom GNU Radio Sink. If you would like to transmit from a different port than the default, you will need to update the `get_antenna_port` function in challengectl.py to select a different port for the device. There is an example in that function, selecting TX2 for one of the RFHS bladeRF devices with a physically broken TX1 port. Setting transmit ports via configuration files will likely be in a future iteration of challengectl, but for now changing the transmit port requires updating the code.
 
+## Avoid Frequencies File
+The avoid frequencies file is a new line delimited JSON file, with each line containing one JSON object with a frequency range or channel to avoid. The purpose of defining frequency ranges or channels in this file, and passing the file into `challengectl.py` with the `-a` flag is to prevent challengectl from transmitting in frequency ranges that you would like to avoid transmitting on. These frequencies could be avoided by selecting a different frequency in the flags file, but using the avoid frequencies file will also prevent transmitting on these channels when you are using a frequency range, such as ham_440 for a challenge in the flags file.
+
+Avoid frequencies are defined as either channels or frequency ranges. Channels have a center frequency and a bandwidth, both specified in Hz. Frequency ranges have a lower frequency, and an upper frequency, also specified in Hz.
+
+The `avoidfreqs.txt.example` file has examples of both channels and ranges. To avoid a channel, such as the Amateur Radio 2M FM Simplex channel, add a line to your avoid frequencies file like this:
+```
+{"name": "2M FM Simplex", "type": "channel", "center": 146520000, "bandwidth": 10000}
+```
+In the examples file, there are also examples of ranges, such as the Amateur Radio 2M SSB Calling frequency. The SSB frequencies to avoid are defined as ranges because it may be easier to think of the SSB frequencies using the dial frequency as the lower frequency (for USB) and adding the bandwith to the dial frequency to get the upper frequency of the range to avoid. An example of a frequency range to avoid in the avoid frequencies file is below:
+```
+{"name": "2M SSB Calling", "type": "range", "lower_freq": 144200000, "upper_freq": 144203000}
+```
+
 ### Supported Devices
 Any transmit capable device that can be used to transmit using a GNURadio gr-osmosdr sink *should* work, however only the following devices have been tested so far.
 - [Nuand bladeRF 2.0](https://www.nuand.com/bladerf-2-0-micro/)

avoidfreqs.txt.example

@@ -0,0 +1,4 @@
+{"name": "2M FM Simplex", "type": "channel", "center": 146520000, "bandwidth": 10000}
+{"name": "70CM FM Simplex", "type": "channel", "center": 446000000, "bandwidth": 10000}
+{"name": "2M SSB Calling", "type": "range", "lower_freq": 144200000, "upper_freq": 144203000}
+{"name": "70CM SSB Calling", "type": "range", "lower_freq": 432100000, "upper_freq": 432103000}

challengectl.py

@@ -13,6 +13,8 @@
 import string
 import argparse
 import subprocess
+import math
+import json
 
 from challenges import ask, cw, usb_tx, nbfm, spectrum_paint, pocsagtx_osmocom, lrs_pager, lrs_tx
 from challenges.gotenna import gotenna_pro_tx_bladerf, gotenna_pro_tx_usrp
@@ -277,6 +279,117 @@ def fire_gotenna_pro(self, device_string, flag_q, device_q, *flag_args):
         if(replaceinqueue != False):
             flag_q.put(flag_args[0])
 
+class AvoidFreq():
+    """Defines a frequency range to avoid."""
+
+    def __init__(self, name, lower_freq, upper_freq):
+        self.name = name
+        if(lower_freq == upper_freq):
+            # Error, frequencies must be different
+            print("Error: lower and upper frequencies for avoid freq must be different")
+
+        if(lower_freq > upper_freq):
+            # Reverse the lower and upper freqs if they were entered in reverse order
+            self.lower_avoid_freq = upper_freq
+            self.upper_avoid_freq = lower_freq
+        else:
+            self.lower_avoid_freq = lower_freq
+            self.upper_avoid_freq = upper_freq
+
+    def __str__(self):
+        return "Name: {}, {}-{}, Center: {}, Bandwidth: {}".format(self.name, self.lower_avoid_freq, self.upper_avoid_freq, self.get_center_freq(), self.get_bandwidth())
+
+    def is_freq_range_ok(self, lower_freq, upper_freq):
+        """Checks if a frequency range avoids the lower and upper bounds of this AvoidFreq object."""
+
+        # Check if either the lower or upper frequencies of the range are within the AvoidFreq range.
+        is_lower_freq_in_range = self.lower_avoid_freq <= lower_freq <= self.upper_avoid_freq
+        is_upper_freq_in_range = self.lower_avoid_freq <= upper_freq <= self.upper_avoid_freq
+
+        # Check if the lower and upper frequencies are both below or above the AvoidFreq range.
+        is_below_range = lower_freq < self.lower_avoid_freq and upper_freq < self.lower_avoid_freq
+        is_above_range = lower_freq > self.upper_avoid_freq and upper_freq > self.upper_avoid_freq
+
+        # Check if the requested range fully encompasses the AvoidFreq range
+        covers_entire_range = lower_freq <= self.lower_avoid_freq and upper_freq >= self.upper_avoid_freq
+
+        # If either frequency is in the avoid range, return False
+        if(is_lower_freq_in_range == True or is_upper_freq_in_range == True):
+            return False
+        elif(covers_entire_range == True):
+            # Return false if the requested range covers the entire AvoidFreq range.
+            return False
+        elif(is_below_range == True or is_above_range == True):
+            # Return True if the requested range is entirely below or above the AvoidFreq range
+            return True
+        else:
+            # Default return False to block the requested frequency range.
+            print("WARNING: Default AvoidFreq Block: {}-{}".format(lower_freq, upper_freq))
+            return False
+
+    def is_channel_ok(self, center_freq, bandwidth):
+        channel_range = AvoidFreq.channel_to_range(center_freq, bandwidth)
+        lower_freq = channel_range[0]
+        upper_freq = channel_range[1]
+        return self.is_freq_range_ok(lower_freq, upper_freq)
+
+    def get_center_freq(self):
+        """Returns the center frequency of an AvoidFreq object."""
+        return ((self.lower_avoid_freq + self.upper_avoid_freq) / 2)
+
+    def get_bandwidth(self):
+        """Returns the bandwidth of an AvoidFreq object."""
+        return (self.upper_avoid_freq - self.lower_avoid_freq)
+
+    def channel_to_range(center_freq, bandwidth):
+        """Converts a channel with a center frequency and bandwidth to a tuple representing the frequency range."""
+        lower_freq = center_freq - math.ceil(bandwidth / 2)
+        upper_freq = center_freq + math.ceil(bandwidth / 2)
+        return (lower_freq, upper_freq)
+
+    def read_avoid_freqs(avoid_freqs_file):
+        """Read file with frequencies to avoid, specified either with channels with a center frequency and bandwidth, or a start and stop frequency range."""
+        avoid_freqs = []
+        # Read and parse the avoid frequencies file
+        with open(avoid_freqs_file) as a:
+            avoid_freq_list = []
+            raw_avoid_freqs = a.readlines()
+            for line in raw_avoid_freqs:
+                avoid_freq_list.append(json.loads(line))
+
+        # Create an AvoidFreq object for each item in the file.
+        for avoidfreqitem in avoid_freq_list:
+            name = avoidfreqitem['name']
+            avoidtype = avoidfreqitem['type']
+            # Channels are defined by a center frequency and a bandwidth
+            if(avoidtype == "channel"):
+                avoidcenterfreq = avoidfreqitem['center']
+                avoidbandwidth = avoidfreqitem['bandwidth']
+                avoidRange = AvoidFreq.channel_to_range(avoidcenterfreq, avoidbandwidth)
+                avoidlowerfreq = avoidRange[0]
+                avoidupperfreq = avoidRange[1]
+                avoidItem = AvoidFreq(name, avoidlowerfreq, avoidupperfreq)
+                avoid_freqs.append(avoidItem)
+            # Ranges are defined by a lower frequency and an upper frequency.
+            elif(avoidtype == "range"):
+                avoidlowerfreq = avoidfreqitem['lower_freq']
+                avoidupperfreq = avoidfreqitem['upper_freq']
+                avoidItem = AvoidFreq(name, avoidlowerfreq, avoidupperfreq)
+                avoid_freqs.append(avoidItem)
+            else:
+                print("Invalid Avoid Type: {}".format(avoidtype))
+
+        return avoid_freqs
+
+    def check_channel_avoid_freqs(avoidFreqs, center_freq, bandwidth):
+        """Checks a channel against a list of AvoidFreq objects. Returns True if the channel does not conflict with the AvoidFreq ranges."""
+        isFreqOk = True
+        for avoid in avoidFreqs:
+            isFreqOk = avoid.is_channel_ok(center_freq, bandwidth)
+            if(isFreqOk != True):
+                print("txfreq: {}, bandwidth: {}, AvoidFreq: {}, isok: {}".format(center_freq, bandwidth, avoid.name, isFreqOk))
+                return False
+        return isFreqOk
 
 def select_freq(band):
     """Read from frequencies text file, select row that starts with band argument.
@@ -289,6 +402,26 @@ def select_freq(band):
                 freq = randint(int(row[1]), int(row[2]))
                 return((freq, row[1], row[2]))
 
+def get_challenge_bandwidth(challengetype):
+    """Returns the bandwidth for a given challenge type."""
+    if(challengetype == "cw"):
+        return 100
+    elif(challengetype == "nbfm"):
+        return 10000
+    elif(challengetype == "usb"):
+        return 2700
+    elif(challengetype == "ask"):
+        return 600
+    elif(challengetype == "pocsag"):
+        return 9000
+    elif(challengetype == "lrs"):
+        return 12500
+    elif(challengetype == "gotenna_pro"):
+        return 25000
+    else:
+        print("WARNING: Default case reached in get_challenge_bandwidth. challengetype: {}".format(challengetype))
+        return 10000
+
 def select_dvbt(channel):
     with open("dvbt_channels.txt") as f:
         reader = csv.reader(f)
@@ -355,6 +488,7 @@ def argument_parser():
     parser = argparse.ArgumentParser(description="A script to run SDR challenges on multiple SDR devices.")
     parser.add_argument("-f", '--flagfile', help="Flags file")
     parser.add_argument("-d", '--devicefile', help="Devices file")
+    parser.add_argument("-a", '--avoidfreqsfile', help="Avoid Frequencies file")
     parser.add_argument("-v", "--verbose", action="store_true")
     parser.add_argument("-t", "--test", help="Run each challenge once to test flags.", action="store_true")
     return parser
@@ -367,8 +501,15 @@ def main(options=None):
     args = options
     flagfile = args.flagfile
     devicefile = args.devicefile
+    avoidfreqsfile = args.avoidfreqsfile
     verbose = args.verbose
     test = args.test
+
+    # Read and parse the Avoid Frequencies file
+    avoidFreqs = []
+    if(avoidfreqsfile is not None):
+        avoidFreqs = AvoidFreq.read_avoid_freqs(avoidfreqsfile)
+
     global conference
     # Create thread safe FIFO queues for devices and flags
     device_Q = Queue()
@@ -436,29 +577,53 @@ def main(options=None):
                 txfreq = freq_range[0]
                 freq_or_range = str(freq_range[1]) + "-" + str(freq_range[2])
 
-            # Paint waterfall every time during the CTF, or only once when testing
-            if(test != True or challenges_transmitted == 0):
-                print(f"\nPainting Waterfall on {txfreq}\n")
-                # spectrum_paint.main(current_chal[7] * 1000, fetch_device(dev_available))
-                antenna = get_antenna_port(dev_available)
-
-                p = Process(target=spectrum_paint.main, args=(txfreq * 1000, dev_available, antenna))  # , daemon=True)
+            # Convert transmit frequency from kHz to Hz
+            txfreq_hz = txfreq * 1000
+            # Default spectrum paint bandwidth is 2 MHz
+            spectrum_paint_bandwidth = 2000000
+
+            isFreqOk = AvoidFreq.check_channel_avoid_freqs(avoidFreqs, txfreq_hz, spectrum_paint_bandwidth)
+            # Only spectrum paint if the spectrum paint avoids all AvoidFreq ranges
+            if(isFreqOk == True):
+                # Paint waterfall every time during the CTF, or only once when testing
+                if(test != True or challenges_transmitted == 0):
+                    print(f"\nPainting Waterfall on {txfreq}\n")
+                    # spectrum_paint.main(current_chal[7] * 1000, fetch_device(dev_available))
+                    antenna = get_antenna_port(dev_available)
+
+                    p = Process(target=spectrum_paint.main, args=(txfreq * 1000, dev_available, antenna))  # , daemon=True)
+                    p.start()
+                    p.join()
+                    disable_amp(dev_available)
+            else:
+                print("WARNING: Skipped spectrum paint due to Avoid Frequency conflict.")
+
+            challenge_bandwidth = get_challenge_bandwidth(cc_module)
+            challenge_channel_center = txfreq_hz
+            if(cc_module == "usb"):
+                # For the USB flowgraph, the tx freq is not the center of the channel.
+                # Calculate where the center of the transmission will be, and use that to check the AvoidFreq ranges.
+                challenge_channel_center = txfreq_hz + int(challenge_bandwidth / 2)
+            else:
+                challenge_channel_center = txfreq_hz
+            isChallengeFreqOk = AvoidFreq.check_channel_avoid_freqs(avoidFreqs, challenge_channel_center, challenge_bandwidth)
+
+            if(isChallengeFreqOk == True):
+                print(f"\nStarting {cc_name} on {txfreq}")
+                # Create list of challenge module arguments, using txfreq to allow setting random freq here instead of in the challenge module
+                replaceinqueue = True
+                norandsleep = False
+                if(test):
+                    replaceinqueue = False
+                    norandsleep = True
+                challengeargs = [cc_id, cc_flag, cc_modopt1, cc_modopt2, cc_minwait, cc_maxwait, txfreq, replaceinqueue, norandsleep]
+                p = Process(target=getattr(t, "fire_" + cc_module), args=(dev_available, flag_Q, device_Q, challengeargs))
                 p.start()
-                p.join()
-                disable_amp(dev_available)
-            print(f"\nStarting {cc_name} on {txfreq}")
-            # Create list of challenge module arguments, using txfreq to allow setting random freq here instead of in the challenge module
-            replaceinqueue = True
-            norandsleep = False
-            if(test):
-                replaceinqueue = False
-                norandsleep = True
-            challengeargs = [cc_id, cc_flag, cc_modopt1, cc_modopt2, cc_minwait, cc_maxwait, txfreq, replaceinqueue, norandsleep]
-            p = Process(target=getattr(t, "fire_" + cc_module), args=(dev_available, flag_Q, device_Q, challengeargs))
-            p.start()
-            if(test == True):
-                jobs.append(p)
-            challenges_transmitted += 1
+                if(test == True):
+                    jobs.append(p)
+                challenges_transmitted += 1
+            else:
+                print("WARNING: Challenge skipped due to Avoid Frequency conflict.")
             # #we need a way to know if p.start errored or not
             # os.system("echo " + freq_or_range + " > /run/shm/wctf_status/" + current_chal[8] + "_sdr")
             # os.system('''timeout 15 ssh -F /root/wctf/liludallasmultipass/ssh/config -oStrictHostKeyChecking=no -oConnectTimeout=10 -oPasswordAuthentication=no -n scoreboard echo ''' + freq_or_range + " > /run/shm/wctf_status/" + current_chal[8] + "_sdr")

flags.txt.ci

@@ -1,5 +1,5 @@
 testflagsfile,1660312800,1660496400
-0,HAM_DIGITAL_9,flags-archive/defcon33/HD4-BPSK-125-wagon.wav,usb,48000,,60,90,147105,,
+0,HAM_DIGITAL_9,flags-archive/defcon33/HD4-BPSK-125-wagon.wav,usb,48000,,60,90,148105,,
 1,HAM_DIGITAL_6,flags-archive/defcon33/HD1-MT63-500L-sandbox.wav,nbfm,48000,,60,90,443925,,
 2,HAM_MORSE_1,GO TAKE THE HAM EXAM,cw,35,,60,90,ham_900,,
 3,SDR_ASK_1,The Silence of the Lambs was filmed in Pittsburgh Pennsylvania.,ask,,,60,90,ham_440,,