target/fake_trx: initial release of virtual transceiver

This is a set of tools for creating a virtual Um-interface between
OsmocomBB and OsmoBTS. It may be extremely useful for testing and
development of GSM stack, including both sides (MS and BTS). This
software implements OsmoTRX (Osmocom's fork of OpenBTS transceiver)
style clock (CLCK), control (CTRL) and data interfaces. So, OsmoBTS
source code doesn't require any modifications, while for OsmocomBB
you will need to use a new application - trxcon, which can be found
in the 'fixeria/sdr_phy' branch until one is merged to master.

Brief description of available applications:

  - fake_trx.py - main application, that allows to connect both
    OsmocomBB and OsmoBTS without actual RF hardware. Currently
    only a single MS may work with a single BTS.

  - clck_gen.py - a peripheral tool aimed to emulate TDMA frame
    clock generator. Could be used for testing and clock
    synchronization of multiple applications. It should be noted,
    that one relays on generic system timer (via Python), so
    a random clock jitter takes place.

  - ctrl_cmd.py - another peripheral tool, which could be used
    for sending CTRL commands directly in manual mode, and also
    for application fuzzing.

Change-Id: Ib1fb80682002ac85a72fa6abef459a4c44f4ab97

1 files changed, 204 insertions, 0 deletions

diff --git a/src/target/fake_trx/fake_trx.py b/src/target/fake_trx/fake_trx.py
new file mode 100755
index 00000000..c30382b4
--- /dev/null
+++ b/src/target/fake_trx/fake_trx.py
@@ -0,0 +1,204 @@
+#!/usr/bin/env python2
+# -*- coding: utf-8 -*-
+
+# Virtual Um-interface (fake transceiver)
+# OsmocomBB <-> OsmoBTS bridge
+#
+# (C) 2017 by Vadim Yanitskiy <axilirator@gmail.com>
+#
+# All Rights Reserved
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+
+import signal
+import getopt
+import select
+import sys
+
+from ctrl_if_bts import CTRLInterfaceBTS
+from ctrl_if_bb import CTRLInterfaceBB
+from burst_fwd import BurstForwarder
+
+from udp_link import UDPLink
+from clck_gen import CLCKGen
+
+COPYRIGHT = \
+	"Copyright (C) 2017 by Vadim Yanitskiy <axilirator@gmail.com>\n" \
+	"License GPLv2+: GNU GPL version 2 or later " \
+	"<http://gnu.org/licenses/gpl.html>\n" \
+	"This is free software: you are free to change and redistribute it.\n" \
+	"There is NO WARRANTY, to the extent permitted by law.\n"
+
+class Application:
+	# Application variables
+	bts_addr = "127.0.0.1"
+	bb_addr = "127.0.0.1"
+	bts_base_port = 5700
+	bb_base_port = 5703
+
+	def __init__(self):
+		self.print_copyright()
+		self.parse_argv()
+
+		# Set up signal handlers
+		signal.signal(signal.SIGINT, self.sig_handler)
+
+	def run(self):
+		# Init TRX CTRL interface for BTS
+		self.bts_ctrl = CTRLInterfaceBTS(self.bts_addr,
+			self.bts_base_port + 101, self.bts_base_port + 1)
+
+		# Init TRX CTRL interface for BB
+		self.bb_ctrl = CTRLInterfaceBB(self.bb_addr,
+			self.bb_base_port + 101, self.bb_base_port + 1)
+
+		# BTS <-> BB burst forwarding
+		self.bts_data = UDPLink(self.bts_addr,
+			self.bts_base_port + 102, self.bts_base_port + 2)
+		self.bb_data = UDPLink(self.bb_addr,
+			self.bb_base_port + 102, self.bb_base_port + 2)
+		self.burst_fwd = BurstForwarder(self.bts_data, self.bb_data)
+
+		# Share clock between BTS and BB
+		self.bts_clck = UDPLink(self.bts_addr,
+			self.bts_base_port + 100, self.bts_base_port)
+		self.bb_clck = UDPLink(self.bb_addr,
+			self.bb_base_port + 100, self.bb_base_port)
+		self.clck_gen = CLCKGen([self.bts_clck, self.bb_clck])
+		self.clck_gen.start()
+
+		print("[i] Init complete")
+
+		# Enter main loop
+		while True:
+			socks = [self.bts_ctrl.sock, self.bb_ctrl.sock,
+				self.bts_data.sock, self.bb_data.sock]
+
+			# Wait until we get any data on any socket
+			r_event, w_event, x_event = select.select(socks, [], [])
+
+			# Downlink: BTS -> BB
+			if self.bts_data.sock in r_event:
+				self.burst_fwd.bts2bb()
+
+			# Uplink: BB -> BTS
+			if self.bb_data.sock in r_event:
+				self.burst_fwd.bb2bts()
+
+			# CTRL commands from BTS
+			if self.bts_ctrl.sock in r_event:
+				data, addr = self.bts_ctrl.sock.recvfrom(128)
+				self.bts_ctrl.handle_rx(data)
+
+			# CTRL commands from BB
+			if self.bb_ctrl.sock in r_event:
+				data, addr = self.bb_ctrl.sock.recvfrom(128)
+				self.bb_ctrl.handle_rx(data)
+
+	def shutdown(self):
+		print("[i] Shutting down...")
+
+		# Stop clock generator
+		self.clck_gen.stop()
+
+		# Close CLCK interfaces
+		self.bts_clck.shutdown()
+		self.bb_clck.shutdown()
+
+		# Close CTRL interfaces
+		self.bts_ctrl.shutdown()
+		self.bb_ctrl.shutdown()
+
+		# Close DATA interfaces
+		self.bts_data.shutdown()
+		self.bb_data.shutdown()
+
+
+	def print_copyright(self):
+		print(COPYRIGHT)
+
+	def print_help(self, msg = None):
+		s  = " Usage: " + sys.argv[0] + " [options]\n\n" \
+			 " Some help...\n" \
+			 "  -h --help           this text\n\n"
+
+		s += " TRX interface specific\n" \
+			 "  -R --bts-addr       Set BTS remote address (default %s)\n"   \
+			 "  -r --bb-addr        Set BB remote address (default %s)\n"    \
+			 "  -P --bts-base-port  Set BTS base port number (default %d)\n" \
+			 "  -p --bb-base-port   Set BB base port number (default %d)\n"
+
+		print(s % (self.bts_addr, self.bb_addr,
+			self.bts_base_port, self.bb_base_port))
+
+		if msg is not None:
+			print(msg)
+
+	def parse_argv(self):
+		try:
+			opts, args = getopt.getopt(sys.argv[1:],
+				"R:r:P:p:h",
+				["help", "bts-addr=", "bb-addr=",
+					"bts-base-port=", "bb-base-port="])
+		except getopt.GetoptError as err:
+			self.print_help("[!] " + str(err))
+			sys.exit(2)
+
+		for o, v in opts:
+			if o in ("-h", "--help"):
+				self.print_help()
+				sys.exit(2)
+
+			elif o in ("-R", "--bts-addr"):
+				self.bts_addr = v
+			elif o in ("-r", "--bb-addr"):
+				self.bb_addr = v
+
+			elif o in ("-P", "--bts-base-port"):
+				self.bts_base_port = int(v)
+			elif o in ("-p", "--bb-base-port"):
+				self.bb_base_port = int(v)
+
+		# Ensure there is no overlap between ports
+		if self.bts_base_port == self.bb_base_port:
+			self.print_help("[!] BTS and BB base ports should be different")
+			sys.exit(2)
+
+		bts_ports = [
+			self.bts_base_port + 0, self.bts_base_port + 100,
+			self.bts_base_port + 1, self.bts_base_port + 101,
+			self.bts_base_port + 2, self.bts_base_port + 102,
+		]
+
+		bb_ports = [
+			self.bb_base_port + 0, self.bb_base_port + 100,
+			self.bb_base_port + 1, self.bb_base_port + 101,
+			self.bb_base_port + 2, self.bb_base_port + 102,
+		]
+
+		for p in bb_ports:
+			if p in bts_ports:
+				self.print_help("[!] BTS and BB ports overlap detected")
+				sys.exit(2)
+
+	def sig_handler(self, signum, frame):
+		print("Signal %d received" % signum)
+		if signum is signal.SIGINT:
+			self.shutdown()
+			sys.exit(0)
+
+if __name__ == '__main__':
+	app = Application()
+	app.run()