#!/usr/bin/python2.7 # vim: tabstop=4 softtabstop=4 shiftwidth=4 textwidth=80 smarttab expandtab """ * Copyright (C) 2012 Sangoma Technologies Corp. * All Rights Reserved. * * Author(s) * David Yat Sin """ import sys import os import time import logging import re import oswc import sngpy import pyodbc import subprocess import signal import pytables as iptc from pytables.helpers import * import netifaces as ni import errno import ctypes import ipaddr from optparse import OptionParser from threading import Timer mediamon_logger = None def get_mediamon_logger(): global mediamon_logger return mediamon_logger class DBAction(object): @staticmethod def insert(session_id, source_ip, destination_ip, forward_ip, rtp_port): mediamon = get_mediamon() query = """ INSERT into %s (session_id, source_ip, destination_ip, forward_ip, rtp_port) VALUES (?, ?, ?, ?, ?) """ % (mediamon.sessions_table) mediamon.sql_exec(query, session_id, source_ip, destination_ip, forward_ip, rtp_port) @staticmethod def delete(session_id): mediamon = get_mediamon() query = "DELETE from %s WHERE session_id = '%s' " % (mediamon.sessions_table, session_id) mediamon.sql_exec(query) @staticmethod def update(session_id, source_ip, destination_ip, forward_ip, rtp_port): mediamon = get_mediamon() query = "UPDATE %s SET source_ip = '%s', destination_ip = '%s', forward_ip = '%s', rtp_port = '%d' WHERE session_id = '%s' " % (mediamon.sessions_table, source_ip, destination_ip, forward_ip, rtp_port, session_id) mediamon.sql_exec(query) class IPTableAction(object): def __init__(self, mediamon): global mediamon_logger self._table_nat = { 4: iptc.Table("nat", autocommit=False), 6: iptc.Table6("nat", autocommit=False) } self._table_filter = { 4: iptc.Table("filter", autocommit=False), 6: iptc.Table6("filter", autocommit=False) } self._prerouting_chain = { 4: iptc.Chain(self._table_nat[4], mediamon.prerouting_chain_name[4]), 6: iptc.Chain(self._table_nat[6], mediamon.prerouting_chain_name[6]) } self._postrouting_chain = { 4: iptc.Chain(self._table_nat[4], mediamon.postrouting_chain_name[4]), 6: iptc.Chain(self._table_nat[6], mediamon.postrouting_chain_name[6]) } self._forward_chain = { 4: iptc.Chain(self._table_filter[4], mediamon.forward_chain_name[4]), 6: iptc.Chain(self._table_filter[6], mediamon.forward_chain_name[6]) } self._rule_class = { 4: iptc.Rule, 6: iptc.Rule6 } self._mediamon = mediamon self._logger = mediamon._logger mediamon_logger = self._logger self.num_subchains = 0 self.subchain_size = 0 #hashtables of subchains table indexed by their first port self._pre_subchains = { 4: dict(), 6: dict() } self._post_subchains = { 4: dict(), 6: dict() } @iptc_command(get_mediamon_logger) def enable_loopback_accept(self, netif): yield self._table_filter[4] self._logger.debug("Enabling loopback accept for interface %s" % netif) # allow all packets from sngdspx interfaces to reach 127.x.x.x subnet input_chain = iptc.Chain(self._table_filter[4], "INPUT") rule = iptc.Rule() rule.protocol = "udp" rule.in_interface = netif rule.dst = "127.0.0.0/8" rule.target = iptc.Target(rule, "ACCEPT") input_chain.insert_rule(rule) # allow all packets from 127.x.x.x subnet to reach sngdspx interfaces output_chain = iptc.Chain(self._table_filter[4], "OUTPUT") rule = iptc.Rule() rule.protocol = "udp" rule.out_interface = netif rule.src = "127.0.0.0/8" rule.target = iptc.Target(rule, "ACCEPT") output_chain.insert_rule(rule) @iptc_command(get_mediamon_logger) def disable_loopback_accept(self, netif): yield self._table_filter[4] self._logger.debug("Disabling loopback accept for interface %s " % netif) input_chain = iptc.Chain(self._table_filter[4], "INPUT") rule = iptc.Rule() rule.protocol = "udp" rule.in_interface = netif rule.dst = "127.0.0.0/8" rule.target = iptc.Target(rule, "ACCEPT") input_chain.delete_rule(rule) output_chain = iptc.Chain(self._table_filter[4], "OUTPUT") rule = iptc.Rule() rule.protocol = "udp" rule.out_interface = netif rule.src = "127.0.0.0/8" rule.target = iptc.Target(rule, "ACCEPT") output_chain.delete_rule(rule) def insert_chain(self, table, chain): #creates a new chain within table if it does not exist - do not retry for ch in table.chains: if ch.name == chain.name: self._logger.debug("Chain %s already exists" % chain.name) return try: table.create_chain(chain) except (iptc.IPTCError, iptc.XTablesError), e: self._logger.error("Failed to insert chain %s in table %s: %s" % (chain.name, table.name, str(e))) raise @iptc_command(get_mediamon_logger) def setup_prerouting_chain(self, version): yield self._table_nat[version] self.insert_chain(self._table_nat[version], self._prerouting_chain[version]) input_chain = iptc.Chain(self._table_nat[version], "PREROUTING") rule = self._rule_class[version]() iptc.Target(rule, self._mediamon.prerouting_chain_name[version]) input_chain.insert_rule(rule) @iptc_command(get_mediamon_logger) def setup_postrouting_chain(self, version): table = self._table_nat[version] yield table self.insert_chain(table, self._postrouting_chain[version]) input_chain = iptc.Chain(table, "POSTROUTING") rule = self._rule_class[version]() iptc.Target(rule, self._mediamon.postrouting_chain_name[version]) input_chain.insert_rule(rule) @iptc_command(get_mediamon_logger) def setup_forward_chain(self, version): table = self._table_filter[version] yield table if table.is_chain(self._forward_chain[version]) is False: table.create_chain(self._forward_chain[version]) input_chain = iptc.Chain(table, "FORWARD") rule = self._rule_class[version]() iptc.Target(rule, self._mediamon.forward_chain_name[version]) input_chain.insert_rule(rule) def setup_chains(self): for ipversion in [ 4, 6 ]: self.setup_prerouting_chain(ipversion) self.setup_postrouting_chain(ipversion) self.setup_forward_chain(ipversion) @iptc_command(get_mediamon_logger) def flush_chains(self, log_error=True): for ipversion in [ 4, 6 ]: yield self._table_filter[ipversion] yield self._table_nat[ipversion] for ipversion in [ 4, 6 ]: self._prerouting_chain[ipversion].flush() self._postrouting_chain[ipversion].flush() for ipversion in [ 4, 6 ]: self._forward_chain[ipversion].flush() def remove_references(self, chain, target): for r in chain.rules: if (str(r.target.name) == target): chain.delete_rule(r) @iptc_command(get_mediamon_logger) def remove_rules(self): for ipversion in [ 4, 6 ]: yield self._table_nat[ipversion] yield self._table_filter[ipversion] for ipversion in [ 4, 6 ]: self.remove_references(iptc.Chain(self._table_nat[ipversion], 'PREROUTING'), self._mediamon.prerouting_chain_name[ipversion]) self.remove_references(iptc.Chain(self._table_nat[ipversion], 'POSTROUTING'), self._mediamon.postrouting_chain_name[ipversion]) for ipversion in [ 4, 6 ]: self.remove_references(iptc.Chain(self._table_filter[ipversion], 'FORWARD'), self._mediamon.forward_chain_name[ipversion]) @iptc_command(get_mediamon_logger) def remove_chains(self): for ipversion in [ 4, 6 ]: yield self._table_nat[ipversion] yield self._table_filter[ipversion] for ipversion in [ 4, 6 ]: self.delete_chain(self._table_nat[ipversion], self._postrouting_chain[ipversion]) self.delete_chain(self._table_nat[ipversion], self._prerouting_chain[ipversion]) for ipversion in [ 4, 6 ]: self.delete_chain(self._table_filter[ipversion], self._forward_chain[ipversion]) def _get_pre_subchain(self, port, ipversion): for port_start, subchain in self._pre_subchains[ipversion].iteritems(): if port < port_start: continue elif port > (port_start + self.subchain_size): continue else: return subchain return None def _get_post_subchain(self, port, ipversion): for port_start, subchain in self._post_subchains[ipversion].iteritems(): if port < port_start: continue elif port > (port_start + self.subchain_size): continue else: return subchain return None @iptc_command(get_mediamon_logger) def create_pre_subchain(self, portrange_start, portrange_stop, version): subchain_name = "mediamon-pre-{!s}-{!s}".format(portrange_start , portrange_stop) self._logger.debug("Creating subchain: {}".format(subchain_name)) table = self._table_nat[version] yield table subchain = iptc.Chain(table, subchain_name) table.create_chain(subchain) rule = self._rule_class[version](protocol="udp") iptc.Match(rule, "udp", dport="{!s}:{!s}".format(portrange_start, portrange_stop)) iptc.Target(rule, subchain_name) self._prerouting_chain[version].insert_rule(rule) self._pre_subchains[version][portrange_start] = subchain @iptc_command(get_mediamon_logger) def create_post_subchain(self, portrange_start, portrange_stop, ipversion): subchain_name = "mediamon-post-{!s}-{!s}".format(portrange_start , portrange_stop) self._logger.debug("Creating subchain: {}".format(subchain_name)) table = self._table_nat[ipversion] yield table subchain = iptc.Chain(table, subchain_name) table.create_chain(subchain) rule = self._rule_class[ipversion](protocol="udp") iptc.Match(rule, "udp", sport="{!s}:{!s}".format(portrange_start, portrange_stop)) iptc.Target(rule, subchain_name) self._postrouting_chain[ipversion].insert_rule(rule) self._post_subchains[ipversion][portrange_start] = subchain def delete_chain(self, table, chain): for r in chain.rules: chain.delete_rule(r) table.delete_chain(chain) @iptc_command(get_mediamon_logger) def remove_subchains(self, ipversion): yield self._table_nat[ipversion] for key, subchain in self._pre_subchains[ipversion].iteritems(): self.delete_chain(self._table_nat[ipversion], subchain) for key, subchain in self._post_subchains[ipversion].iteritems(): self.delete_chain(self._table_nat[ipversion], subchain); def _create_forward_rule(self, portdir, port_range_start, port_range_stop, ipdir, ip, ipversion): rule = self._rule_class[ipversion](protocol="udp") setattr(rule, ipdir, ip) match = iptc.Match(rule, "udp") setattr(match, portdir, "%d:%d" % (port_range_start, port_range_stop)) iptc.Target(rule, "ACCEPT") chain = self._forward_chain[ipversion] chain.insert_rule(rule) @iptc_command(get_mediamon_logger) def enable_forward(self, port_range_start, port_range_stop, ip, ipversion): yield self._table_filter[ipversion] self._create_forward_rule('dport', port_range_start, port_range_stop, 'dst', ip, ipversion) self._create_forward_rule('sport', port_range_start, port_range_stop, 'src', ip, ipversion) @iptc_command(get_mediamon_logger) def enable_outbound_drop(self, ip_address, ipversion): yield self._table_nat rule = self._rule_class[ipversion](src = ip_address, protocol = "udp") iptc.Target(rule, "DROP") self._prerouting_chain[ipversion].insert_rule(rule) @iptc_command(get_mediamon_logger) def enable_inbound_drop(self, netif, port_range_start, port_range_stop, ipversion): yield self._table_nat[ipversion] rule = self._rule_class[ipversion](in_interface = netif, protocol = "udp") iptc.Match(rule, "udp", dport="{!s}:{!s}".format(port_range_start, port_range_stop)) iptc.Target(rule, "DROP") self._prerouting_chain[ipversion].insert_rule(rule) @iptc_command(get_mediamon_logger) def enable_outbound_snat(self, netif, from_ip_address, to_ip_address): to_ip = ipaddr.IPAddress(to_ip_address) ni_version = ni.AF_INET if to_ip.version == 4 else ni.AF_INET6 if ni_version not in ni.ifaddresses(netif): iptc_abort() netif_ips = [] for address in ni.ifaddresses(netif)[ni_version]: if (not address['addr'].lower().startswith("fe80")): netif_ips.append(address['addr']) if to_ip_address not in netif_ips: iptc_abort() chain = self._postrouting_chain[to_ip.version] yield chain.table rule = self._rule_class[to_ip.version](protocol="udp", src=from_ip_address) iptc.Target(rule, "SNAT", to_source = (self._mediamon.options.global_src_ip \ if self._mediamon.options.global_src_ip else to_ip_address)) chain.insert_rule(rule) @iptc_command(get_mediamon_logger) def enable_inbound_dnat(self, netif, to_ip_address, from_ip_address, port_range_start, port_range_stop): to_ip = ipaddr.IPAddress(to_ip_address) ni_version = ni.AF_INET if to_ip.version == 4 else ni.AF_INET6 if ni_version not in ni.ifaddresses(netif): iptc_abort() netif_ips = [] for address in ni.ifaddresses(netif)[ni_version]: if (not address['addr'].lower().startswith("fe80")): netif_ips.append(address['addr']) if from_ip_address not in netif_ips: logger.debug('address {!s} not in interface {!s} or invalid, ignoring'.format(from_ip_address, netif)) iptc_abort() chain = self._prerouting_chain[to_ip.version] yield chain.table rule = self._rule_class[to_ip.version](in_interface=netif, protocol="udp", dst=from_ip_address) iptc.Match(rule, "udp", dport="{!s}:{!s}".format(port_range_start, port_range_stop)) iptc.Target(rule, "DNAT", to_destination=to_ip_address) chain.insert_rule(rule) @iptc_command(get_mediamon_logger) def enable_loop_dnat(self, netif, external_netif, dest_ip_address, to_ip_address, port_range_start, port_range_stop): ip = ipaddr.IPAddress(to_ip_address) addrfamily = { 4: ni.AF_INET, 6: ni.AF_INET6 }[ip.version] if addrfamily not in ni.ifaddresses(external_netif): iptc_abort() netif_ips = [] for address in ni.ifaddresses(external_netif)[addrfamily]: if (not address['addr'].lower().startswith("fe80")): netif_ips.append(address['addr']) if dest_ip_address not in netif_ips: logger.debug('address {!s} not in interface {!s} or invalid, ignoring'.format(dest_ip_address, external_netif)) iptc_abort() chain = self._prerouting_chain[ip.version] yield chain.table rule = self._rule_class[ip.version](in_interface=netif, protocol="udp") addr = dest_ip_address if ip.version == 6: sindex = addr.find("%") if sindex > 0: addr = addr[:sindex] rule.dst = addr iptc.Match(rule, "udp", dport = "{!s}:{!s}".format(port_range_start, port_range_stop)) iptc.Target(rule, "DNAT", to_destination=to_ip_address) self._logger.debug("inserting DNAT rule on PREROUTING (netif=%s, addr=%s, range=%d-%d, to_ip=%s)" % \ (netif, addr, port_range_start, port_range_stop, to_ip_address)) chain.insert_rule(rule) yield iptc_commit() def _create_accept(self, forward_ip, port, ipversion): rule = self._rule_class[ipversion](protocol="udp", src=forward_ip) iptc.Match(rule, "udp", sport="{!s}:{!s}".format(port, port + 1)) iptc.Target(rule, "ACCEPT") return rule @iptc_command(get_mediamon_logger) def _disable_accept(self, forward_ip, port, ipversion): rule = self._create_accept(forward_ip, port, ipversion) chain = self._prerouting_chain[ipversion] yield chain.table chain.delete_rule(rule) @iptc_command(get_mediamon_logger) def _enable_accept(self, forward_ip, port, ipversion): rule = self._create_accept(forward_ip, port, ipversion) chain = self._prerouting_chain[ipversion] yield chain.table chain.insert_rule(rule) def _create_snat(self, destination_ip, port, ipversion): rule = self._rule_class[ipversion](protocol = "udp") iptc.Match(rule, "udp", sport="{!s}:{!s}".format(port, port)) iptc.Target(rule, "SNAT", to_source="{!s}:{!s}".format(destination_ip, port)) return rule @iptc_command(get_mediamon_logger) def _enable_snat(self, destination_ip, port, ipversion): subchain = self._get_post_subchain(port, ipversion) yield subchain.table rule_rtp = self._create_snat(destination_ip, port, ipversion) rule_rtcp = self._create_snat(destination_ip, port + 1, ipversion) if subchain is None: self._logger.error("Failed to get subchain for port %d" % port) iptc_abort() subchain.insert_rule(rule_rtp) subchain.insert_rule(rule_rtcp) @iptc_command(get_mediamon_logger) def _disable_snat(self, destination_ip, port, ipversion): subchain = self._get_post_subchain(port, ipversion) yield subchain.table rule_rtp = self._create_snat(destination_ip, port, ipversion) rule_rtcp = self._create_snat(destination_ip, port + 1, ipversion) if subchain is None: self._logger.error("Failed to get subchain for port %d" % port) iptc_abort() subchain.delete_rule(rule_rtp) subchain.delete_rule(rule_rtcp) def _create_dnat(self, destination_ip, forward_ip, port, ipversion): """ When the host is behind NAT and we have not learnt the host IP, we cannot use source_ip yet because it could be a natted IP and currently we do not receive events from vocallo when the real source UDP IP is detected and have no ability to reset the iptables rule Furthermore, if the intention is to protect against possible media corruption by someone eavesdropping our SIP packets, if they know the destination port, they also know the source IP and could easily forge the src IP and we would be screwed anyways, so no much value-added by filtering by source IP """ rule = self._rule_class[ipversion](protocol="udp", dst=destination_ip) iptc.Match(rule, "udp", dport="{!s}:{!s}".format(port, port)) iptc.Target(rule, "DNAT", to_destination="{ip!s}:{port!s}".format(ip=forward_ip, port=port)) return rule @iptc_command(get_mediamon_logger) def _enable_dnat(self, destination_ip, forward_ip, port, ipversion): subchain = self._get_pre_subchain(port, ipversion) yield subchain.table rule_rtp = self._create_dnat(destination_ip, forward_ip, port, ipversion) rule_rtcp = self._create_dnat(destination_ip, forward_ip, port + 1, ipversion) if subchain is None: self._logger.error("Failed to get subchain for port %d" % port) iptc_abort() subchain.insert_rule(rule_rtp); subchain.insert_rule(rule_rtcp); @iptc_command(get_mediamon_logger) def _disable_dnat(self, destination_ip, forward_ip, port, ipversion): subchain = self._get_pre_subchain(port, ipversion) yield subchain.table rule_rtp = self._create_dnat(destination_ip, forward_ip, port, ipversion) rule_rtcp = self._create_dnat(destination_ip, forward_ip, port + 1, ipversion) if subchain is None: self._logger.error("Failed to get subchain for port %d" % port) iptc_abort() subchain.delete_rule(rule_rtp); subchain.delete_rule(rule_rtcp); def enable(self, destination_ip, forward_ip, rtp_port): """ Note that we used to use the source_ip to setup the DNAT rules, but we've decided there is no point as the source ip can easily be spoofed for UDP packets so it does not really provide a significant security improvement, and it does harm when working behind NAT, because the source_ip that we receive may be NATed, so our rule will not work because the NAT will change the source IP to something different We could use a separate event when the real IP is learnt later on (from the RTP stream), but, what's the point? no added security but just extra complexity and inconvenience """ """ Setup a pair of SNAT rules to change the source address of the internal media interfaces (forward_ip) to the public external IP (destination_ip) The rules are set based on the UDP source port Then we also need to accept the traffic coming from the media interface because by default we drop everything coming from media interfaces to avoid kernel connection tracking to kicking in before we have a chance to setup the SNAT/DNAT rules """ ipversion = 6 if destination_ip.find(':') != -1 else 4 forward_ipversion = 6 if forward_ip.find(':') != -1 else 4 if ipversion != forward_ipversion: self._logger.error('Mismatched address family for "{!s}" (ipv{!s}) and "{!s}" (ipv{!s})'.format(destination_ip, ipversion, forward_ip, forward_ipversion)) self._enable_snat(destination_ip, rtp_port, ipversion) self._enable_accept(forward_ip, rtp_port, ipversion); """ Setup a pair of DNAT rules to change the destination address of packets coming from the external interfaces (not a media interface) to route them (ip_forward) to the proper media interface IP The rules are set based on the UDP destination port (which identifies the media interface) and also based on destination IP, which should add a bit more of efficiency because the rule will only be attached to a given particular destination IP (which belongs to a given network interface) """ self._enable_dnat(destination_ip, forward_ip, rtp_port, ipversion) return True def disable(self, destination_ip, forward_ip, rtp_port): """ Tear down all the rules we created during enable() """ ipversion = 6 if destination_ip.find(':') != -1 else 4 forward_ipversion = 6 if forward_ip.find(':') != -1 else 4 if ipversion != forward_ipversion: self._logger.error('Mismatched address family for "{!s}" (ipv{!s}) and "{!s}" (ipv{!s})'.format(destination_ip, ipversion, forward_ip, forward_ipversion)) self._disable_snat(destination_ip, rtp_port, ipversion) self._disable_accept(forward_ip, rtp_port, ipversion) self._disable_dnat(destination_ip, forward_ip, rtp_port, ipversion) return True class MediaSessionListener(oswc.EventListener): def __init__(self, logger): self._logger = logger super(oswc.EventListener, self).__init__() def on_event(self, event): mediamon = get_mediamon() ip_action = get_ip_action() self._logger.debug("Received media session event %s\n" % event) session_id = event.get_header("session-id") status = event.get_header("status") source_ip = event.get_header("source-ip") destination_ip = event.get_header("destination-ip") forward_ip = event.get_header("forward-ip") rtp_port = eval(event.get_header("rtp-port")) if status == 'enable': mediamon._logger.debug("Enabling session %s,%s,%s,%s,%d" % (session_id, source_ip, destination_ip, forward_ip, rtp_port)) DBAction.insert(session_id, source_ip, destination_ip, forward_ip, rtp_port) ip_action.enable(destination_ip, forward_ip, rtp_port) else: mediamon._logger.debug("Disabling session %s,%s,%s,%s,%d" % (session_id, source_ip, destination_ip, forward_ip, rtp_port)) DBAction.delete(session_id) ip_action.disable(destination_ip, forward_ip, rtp_port) class MediaMonitor(sngpy.DBService): _service_name = 'mediamon' def __init__(self, logger): self._logger = logger self._dbconf = dict() self._dbconf_params = ['connection-string', 'table-prefix'] self._dbconn = None self._swconf = dict() self._swconf_params = ['connection-string'] self._conf = dict() self._conf_params = ['num-subchains', 'dynamic-rules', 'enable-heartbeat'] self.sched = None self._external_interfaces = ni.interfaces() self._dynamic_rules = False self._enable_heartbeat = False self._sleep_interval = 1 self.modules_table = '' self.sessions_table = '' self.map_table = '' self._access_interfaces = [] self._external_interfaces = [] self.sched = None self._oswc_conn = None self.sysctl_bin = '/sbin/sysctl' self.conntrack_bin = '/usr/sbin/conntrack' self.sngtc_bin = '/usr/local/nsc/bin/sngtc_tool' self.prerouting_chain_name = { 4: 'mediamon_prerouting', 6: 'mediamon_prerouting6' } self.postrouting_chain_name = { 4: 'mediamon_postrouting', 6: 'mediamon_postrouting6' } self.forward_chain_name = { 4: 'mediamon_forward', 6: 'mediamon_forward6' } self._dynamic_rules = False self._external_interfaces = ni.interfaces() self._null = None self._conntrack_udp_timeout = 10 self._conntrack_udp_timeout_stream = 40 formatter = logging.Formatter(self._logformat) console_handler = logging.StreamHandler() console_handler.setFormatter(formatter) self._logger.addHandler(console_handler) self.sched = sngpy.Scheduler(time.time, time.sleep) self._null = open(os.devnull, 'w') # custom MediaMon options should be added here via parser.add_option() self.parser = OptionParser() self.parser.add_option("", "--init-database", action="store_true", dest="init_database", help="Initialize Database") self.parser.add_option("", "--show-status", action="store_true", dest="show_status", help="Show current sessions") self.parser.add_option("", "--global-src-ip", action="store", dest="global_src_ip", help="Set a global source IP ignoring interface addresses") super(MediaMonitor, self).__init__() # handle sighup to reload rules and block/unblock objects self.reload = False def sigreload(signum, frame): mediamon = get_mediamon() mediamon.reload = True signal.signal(signal.SIGHUP, sigreload) def _parse_db(self, params): self._logger.debug("Reading database parameters") for p in params: self._logger.debug("database: %s=%s" % (p.attrib['name'], p.attrib['value'])) if p.attrib['name'] in self._dbconf_params: self._dbconf[p.attrib['name']] = p.attrib['value'] else: raise ValueError, "Unknown database XML parameter %s" % p.attrib['name'] if 'connection-string' not in self._dbconf: raise ValueError, "Missing database connection-string XML parameter" if 'table-prefix' not in self._dbconf: self._dbconf['table-prefix'] = '' self.modules_table = self._dbconf['table-prefix'] + "media_modules" self.map_table = self._dbconf['table-prefix'] + "media_map" self.sessions_table = self._dbconf['table-prefix'] + "media_sessions" def _parse_switch(self, params): self._logger.debug("Reading switch parameters") for p in params: self._logger.debug("switch: %s=%s" % (p.attrib['name'], p.attrib['value'])) if p.attrib['name'] in self._swconf_params: self._swconf[p.attrib['name']] = p.attrib['value'] else: raise ValueError, "Unknown switch XML parameter %s" % p.attrib['name'] if 'connection-string' not in self._swconf: raise ValueError, "Missing switch connection-string XML parameter" def _parse_conf(self, params): self._logger.debug("Reading mediamon parameters") for p in params: self._logger.debug("mediamon: %s=%s" % (p.attrib['name'], p.attrib['value'])) if p.attrib['name'] in self._conf_params: self._conf[p.attrib['name']] = p.attrib['value'] else: raise ValueError, "Unknown mediamon XML parameter %s" % p.attrib['name'] if 'num-subchains' not in self._conf: raise ValueError, "Missing mediamon num-subchains XML parameter" if 'dynamic-rules' in self._conf: self._dynamic_rules = self._is_true(self._conf['dynamic-rules']) if 'enable-heartbeat' in self._conf: self._enable_heartbeat = self._is_true(self._conf['enable-heartbeat']) def configure(self): ret = True try: tree = super(MediaMonitor,self).configure() conf = tree.find('mediamon') if conf is None: raise ValueError, "Missing configuration" else: params = self._get_params(conf) self._parse_conf(params) dbconf = tree.find('database') if dbconf is None: raise ValueError, "Missing configuration" else: params = self._get_params(dbconf) self._parse_db(params) swconf = tree.find('switch') if swconf is None: raise ValueError, "Missing configuration" else: params = self._get_params(swconf) self._parse_switch(params) if self._dynamic_rules and self.options.global_src_ip: self._logger.warning('--global-src-ip will be ignored when dynamic rules are enabled') if not self._dynamic_rules and self._enable_heartbeat: self._logger.warning('enable-heartbeat will be ignored when dynamic rules are disabled') except: self._logger.critical("Failed to configure service") exc_type, exc_value, exc_traceback = sys.exc_info() self._print_exception(exc_type, exc_value, exc_traceback) ret = False return ret def cmd_exec(self, cmd, errcheck=True): cmd_str = ' '.join(cmd) self._logger.debug("Executing command %s" % (cmd_str)) rc = subprocess.call(cmd, stdout=self._null, stderr=self._null) if errcheck and rc: self._logger.error("Command %s returned %d" % (cmd_str, rc)) else: self._logger.debug("Command %s returned %d" % (cmd_str, rc)) return rc def init_database(self): # connection to the database takes care of initialization try: self._db_connect() return True except: return False def show_status(self): self._db_connect() query = "SELECT * from %s" % (mediamon.sessions_table) cursor = mediamon.sql_exec(query) num_sessions = 0 print "=========================================================================================================================" print "| Session ID \t| Source \t\t| Destination\t\t| Forward \t\t| RTP |" print "=========================================================================================================================" for obj in cursor: print "| %s\t| %15s\t| %15s\t| %15s\t| %05s\t|" % (obj.session_id, obj.source_ip, obj.destination_ip, obj.forward_ip, obj.rtp_port) num_sessions = num_sessions + 1 print "=========================================================================================================================" print "| Number of sessions:%05d |" %num_sessions print "=========================================================================================================================" return True def _housekeeping(self): self.sched.fast_run() def _connect(self): # Connection loop (wait until we can connect or the daemon is stopped) self._oswc_conn = None connection_string = self._swconf['connection-string'] while self.daemon_alive: self._housekeeping() # connect to the event provider if self._oswc_conn is None: self._oswc_conn = oswc.create_connection(connection_string, logger=self._logger) if self._oswc_conn is None: self._logger.error("Failed to create connection to %s" % (connection_string)) time.sleep(self._sleep_interval) continue if self._enable_heartbeat and self._dynamic_rules: self._oswc_conn.init_heartbeat_checking(self.heartbeat_timeout_handler) listener = MediaSessionListener(self._logger) listener.set_filter(['MEDIA_SESSION']) self._oswc_conn.add_event_listener(listener) if not self._oswc_conn.connect(): self._logger.debug("Failed to connect to %s" % (connection_string)) time.sleep(self._sleep_interval) continue self._logger.info("Connected to %s" % (connection_string)) break def _db_connect(self): self._dbconn = None try: self._dbconn = pyodbc.connect(self._dbconf['connection-string']) except Exception, e: self._logger.error("Failed to connect to the database %s: %s" % (self._dbconf['connection-string'], str(e))) raise # initialize the database self._db_init() def _db_init(self): try: # create media_modules tables dbquery = "CREATE TABLE IF NOT EXISTS `%s` (" """ `id` INT UNSIGNED NOT NULL, `mac_address` TEXT NOT NULL , `access_interface` TEXT NOT NULL , `port_range_start` INT UNSIGNED NOT NULL , `port_range_stop` INT UNSIGNED NOT NULL , PRIMARY KEY (`id`) , UNIQUE INDEX `id_UNIQUE` (`id` ASC) ) ENGINE = InnoDB; """ % (self.modules_table) self.sql_exec(dbquery) # create media_map tables dbquery = "CREATE TABLE IF NOT EXISTS `%s` (" """ `id` INT UNSIGNED NOT NULL, `media_module_id` INT UNSIGNED , `host_rtp_address` TEXT NOT NULL , `module_rtp_address` TEXT NOT NULL , PRIMARY KEY (`id`) , UNIQUE INDEX `id_UNIQUE` (`id` ASC) ) ENGINE = InnoDB; """ % (self.map_table) self.sql_exec(dbquery) # create media_sessions tables dbquery = "CREATE TABLE IF NOT EXISTS `%s` (" """ `id` INT UNSIGNED NOT NULL AUTO_INCREMENT , `session_id` VARCHAR(255) NULL , `source_ip` TEXT NOT NULL , `destination_ip` TEXT NOT NULL , `forward_ip` TEXT NOT NULL , `rtp_port` INT UNSIGNED NOT NULL , PRIMARY KEY (`id`) , UNIQUE INDEX `id_UNIQUE` (`id` ASC) ) ENGINE = InnoDB; """ % (self.sessions_table) self.sql_exec(dbquery) except: self._logger.error("Failed to initialize database") raise def _setup_media_sessions(self): mediamon = get_mediamon() mediamon._logger.info("Enabling existing media sessions") query = "SELECT * from %s" % (mediamon.sessions_table) cursor = mediamon.sql_exec(query) for obj in cursor: mediamon._logger.info("Verifying existing session with uuid: %s" % obj.session_id) cmd_args = "%s local_media_ip remote_media_ip sng_forward_ip sng_forward_port" % (str(obj.session_id)) callback_data = "session_info:%s" % obj.session_id mediamon._oswc_conn.cmd_exec("uuid_getvars", cmd_args, mediamon.cmd_callback, callback_data) def _flush_media_sessions(self): mediamon = get_mediamon() ip_action = get_ip_action() mediamon._logger.info("Disabling existing media sessions") query = "SELECT * from %s" % (mediamon.sessions_table) cursor = mediamon.sql_exec(query) for obj in cursor: ip_action.disable(obj.destination_ip, obj.forward_ip, obj.rtp_port) def _remove_static_rules(self): ip_action = get_ip_action() if self._dynamic_rules: ip_action.remove_subchains() for netif in self._access_interfaces: self._disable_loopback_routing(netif) ip_action.disable_loopback_accept(netif) def _disable_ext_ports(self): query = "SELECT DISTINCT access_interface FROM %s" % (self.modules_table) module_cursor = mediamon.sql_exec(query) for obj in module_cursor: self._logger.debug("Disabling external RJ-45's on %s" % obj.access_interface) cmd = [self.sngtc_bin, '-dev', str(obj.access_interface), '-disable-ext-ports' ] # We currently do not know if we have a D500, therefore we expect this command to fail # with D100, D150 etc self.cmd_exec(cmd, False) def _enable_ext_ports(self): query = "SELECT DISTINCT access_interface FROM %s" % (self.modules_table) module_cursor = mediamon.sql_exec(query) for obj in module_cursor: self._logger.debug("Enabling external RJ-45's on %s" % obj.access_interface) cmd = [self.sngtc_bin, '-dev', str(obj.access_interface), '-enable-ext-ports' ] # We currently do not know if we have a D500, therefore we expect this command to fail # with D100, D150 etc self.cmd_exec(cmd, False) def _setup_static_rules(self): self._access_interfaces = [] self._external_interfaces = [] ip_action = get_ip_action() if self._dynamic_rules: query = "SELECT * FROM %s" % (self.modules_table) else: query = "SELECT * FROM %s INNER JOIN %s ON mediamon_media_modules.id = mediamon_media_map.media_module_id" % (self.map_table, self.modules_table) module_cursor = mediamon.sql_exec(query) # Build list of access_interfaces for obj in module_cursor: if obj.access_interface not in self._access_interfaces: self._access_interfaces.append(obj.access_interface) # All interfaces that are not access interfaces or loopback interface are considered external_interfaces for netif in ni.interfaces(): if (netif not in self._access_interfaces and not netif.startswith("lo") and not netif.startswith("sngdsp")): self._external_interfaces.append(netif) # Enable loopback routing for access interfaces for netif in self._access_interfaces: ip_action.enable_loopback_accept(netif) # Reset the cursor since it was used already on our first pass to build the access interfaces module_cursor = mediamon.sql_exec(query) for obj in module_cursor: if (obj.module_rtp_address.lower().startswith("fe80")): self._logger.debug("Ignoring IPv6 link-local module RTP address %s" % (obj.module_rtp_address)) continue if (obj.host_rtp_address.lower().startswith("fe80")): self._logger.debug("Ignoring IPv6 link-local host RTP address %s" % (obj.host_rtp_address)) continue ipversion = 6 if obj.module_rtp_address.find(':') != -1 else 4 if self._dynamic_rules: self._logger.debug("Adding static rules for %s:%d-%d" % (obj.module_rtp_address, obj.port_range_start, obj.port_range_stop)) ip_action.enable_outbound_drop(obj.module_rtp_address, ipversion) ip_action.enable_forward(obj.port_range_start, obj.port_range_stop, obj.module_rtp_address, ipversion) for netif in self._external_interfaces: ip_action.enable_inbound_drop(netif, obj.port_range_start, obj.port_range_stop, ipversion) ip_action.num_subchains = eval(mediamon._conf['num-subchains']) ip_action.subchain_size = ((obj.port_range_stop - obj.port_range_start) + 1) / ip_action.num_subchains #in terms of functionality it makes no difference to traverse the #list in reverse, but it just makes it easier to debug when looking at iptables for i in reversed(range(ip_action.num_subchains)): portrange_start = (i * ip_action.subchain_size) + obj.port_range_start portrange_stop = portrange_start + ip_action.subchain_size - 1 for ipversion in [ 4, 6 ]: ip_action.create_pre_subchain(portrange_start, portrange_stop, ipversion) ip_action.create_post_subchain(portrange_start, portrange_stop, ipversion) else: self._logger.debug("Adding static fixed rules for %s:%d-%d" % (obj.module_rtp_address, obj.port_range_start, obj.port_range_stop)) ip_action.enable_forward(obj.port_range_start, obj.port_range_stop, obj.module_rtp_address, ipversion) for netif in self._external_interfaces: ip_action.enable_outbound_snat(netif, obj.module_rtp_address, obj.host_rtp_address) ip_action.enable_inbound_dnat(netif, obj.module_rtp_address, obj.host_rtp_address, obj.port_range_start, obj.port_range_stop) for netif in self._access_interfaces: for external_netif in self._external_interfaces: ip_action.enable_loop_dnat(netif, external_netif, obj.host_rtp_address, obj.module_rtp_address, obj.port_range_start, obj.port_range_stop) # Enable local routing for access interfaces for netif in self._access_interfaces: self._enable_loopback_routing(netif) def _flush_conntrack(self): self._logger.debug("Flushing connection tracker") cmd = [self.conntrack_bin, '-D', '-p', 'udp'] # When no conntrack data is available it is normal we get err 1 rc = self.cmd_exec(cmd, False) if rc > 1: self._logger.error("Failed to flush connection tracker") self._logger.debug("Flushing connection tracker for IPv6") cmd = [self.conntrack_bin, '-f', 'ipv6', '-D', '-p', 'udp'] # When no conntrack data is available it is normal we get err 1 rc = self.cmd_exec(cmd, False) if rc > 1: self._logger.error("Failed to flush connection tracker for IPv6: errno %d" % rc) def _setup_conntrack(self): cmd_string = "net.netfilter.nf_conntrack_udp_timeout=%d" % self._conntrack_udp_timeout cmd = [self.sysctl_bin, '-w', cmd_string] self.cmd_exec(cmd) cmd_string = "net.netfilter.nf_conntrack_udp_timeout_stream=%d" % self._conntrack_udp_timeout_stream cmd = [self.sysctl_bin, '-w', cmd_string] self.cmd_exec(cmd) def _enable_loopback_routing(self, netif): # Enable routing of 127.0.0.0/8 addresses on the DSP interfaces # this is needed to route from a local process to nsc (e.g nsg to nsc) # and be able to listen on 127 addresses for the local process and still # send RTP traffic to the DSP interfaces "outside" # note this is only needed for ipv4 as we don't use ipv6 address to # communicate with other internal processes self._logger.debug("Enabling loopback routing for interface %s" % netif) cmd_string = "net.ipv4.conf.%s.route_localnet=1" % netif cmd = [self.sysctl_bin, '-w', cmd_string] self.cmd_exec(cmd) def _disable_loopback_routing(self, netif): self._logger.debug("Disabling loopback routing for interface %s" % netif) cmd_string = "net.ipv4.conf.%s.route_localnet=0" % netif cmd = [self.sysctl_bin, '-w', cmd_string] self.cmd_exec(cmd) def cmd_callback(self, result = None, obj = None): #obj has format = ": " #e.g: session_info: if obj is None: return args_r = re.compile('[:]+') obj_args = args_r.split(obj) #This could be abstracted into a dictionary of callback handlers later on if obj_args[0] == 'session_info': if result[:4] == "-ERR": self._logger.debug("Session with id:%s does not exist anymore" %obj_args[1]) DBAction.delete(obj_args[1]) return vars_r = re.compile('[\n]') variable_lines = vars_r.split(result) source_ip = None destination_ip = None forward_ip = None rtp_port = 0 for variable_line in variable_lines: var_r = re.compile('[:]') variable_pair = var_r.split(variable_line) if variable_pair[0] == "local_media_ip": destination_ip = variable_pair[1] elif variable_pair[0] == "remote_media_ip": source_ip = variable_pair[1] elif variable_pair[0] == "sng_forward_ip": forward_ip = variable_pair[1] elif variable_pair[0] == "sng_forward_port": rtp_port = eval(variable_pair[1]) if source_ip is None or destination_ip is None or forward_ip is None or rtp_port is 0: self._logger.error("Failed to obtain all required parameters for session: %s source_ip:%s destination_ip:%s forward_ip:%s rtp_port:%d" % (obj_args[1], source_ip, destination_ip, forward_ip, rtp_port)) DBAction.delete(obj_args[1]) return ip_action = get_ip_action() ip_action.enable(destination_ip, forward_ip, rtp_port) DBAction.update(obj_args[1], source_ip, destination_ip, forward_ip, rtp_port) else: self._logger.error("Don't how how to handle callback for command:%s" % obj_args[0]) return def heartbeat_timeout_handler(self): self.sched.enter(0, 1, self.reset_server_connection, ()) def reset_server_connection(self): ip_action.flush_chains(False) self._remove_static_rules() ip_action.remove_chains() self._logger.info("Resetting server connection") if self._oswc_conn is not None: self._oswc_conn = None self._connect() self._setup_conntrack() self._disable_ext_ports() ip_action.setup_chains() ip_action.flush_chains() self._setup_static_rules() self._flush_conntrack() self._setup_media_sessions() self._logger.info("%s server connection reset completed" % (self._service_name)) def run(self): super(MediaMonitor, self).run() ip_action = get_ip_action() # connect to the database self._db_connect() # Connect to the event source only if dynamic rules are enabled if self._dynamic_rules: self._connect() self._setup_conntrack() self._disable_ext_ports() ip_action.remove_rules() ip_action.setup_chains() ip_action.flush_chains() self._setup_static_rules() self._flush_conntrack() if self._dynamic_rules: self._setup_media_sessions() self._logger.info("%s is now running" % (self._service_name)) # Event loop while self.daemon_alive: try: if self._dynamic_rules: self._housekeeping() sched_sleep = self.sched.next_event_time_delta() if sched_sleep <= 0: """ You'd think we could run a housekeeping if sched_sleep is zero, but apparently sometimes time goes back and we don't want to call receive_event with 0 as that seems to (oddly) block, just wait one more second to be on the safe side """ sched_sleep = 1; e = self._oswc_conn.receive_event(timeout=sched_sleep*1000) if e is not None and str(e) == 'SERVER_DISCONNECTED': self._logger.info("Server connection lost") self._flush_media_sessions() self._connect() self._setup_media_sessions() else: time.sleep(1) except KeyboardInterrupt: self._logger.info("Stopping %s. User aborted." % (self._service_name)) break except: exc_type, exc_value, exc_traceback = sys.exc_info() self._print_exception(exc_type, exc_value, exc_traceback) time.sleep(self._sleep_interval) self._logger.info("%s is now terminating" % (self._service_name)) # If we are restarting mediamon, we want to keep session information # to maintain the existing calls. if self._dynamic_rules: self._flush_media_sessions() ip_action.flush_chains(False) self._remove_static_rules() ip_action.remove_rules() ip_action.remove_chains() self._enable_ext_ports() logger = logging.getLogger(MediaMonitor._service_name) logger.setLevel(logging.DEBUG) ## main() ## mediamon = None ip_action = None def get_mediamon(): global mediamon return mediamon def get_ip_action(): global mediamon global ip_action if ip_action is None: ip_action = IPTableAction(mediamon) return ip_action try: mediamon = MediaMonitor(logger) if mediamon.options.stop is not None: mediamon.stop() sys.exit(0) if mediamon.options.restart is not None: mediamon.restart() sys.exit(0) if mediamon.options.conf_path is None: # convenient way to retrieve -c option when service is running mediamon.options.conf_path = sngpy.Service.find_conf_path() if mediamon.options.conf_path is None: mediamon.parser.print_help() mediamon.parser.error("-c is required to find the configuration path") sys.exit(1) if mediamon.configure() is False: mediamon.parser.error("Failed to configure daemon using file %s" % mediamon.options.conf_path) sys.exit(1) if mediamon.options.init_database: if mediamon.init_database(): sys.exit(0) else: sys.exit(1) if mediamon.options.show_status: if mediamon.show_status(): sys.exit(0) else: sys.exit(1) # Following options (either start or run) should not be executed if the pid file exists if os.path.exists(mediamon.pidfile): logger.error("Service seems to be running already, pid file %s already exists" % mediamon.pidfile) sys.exit(1) # Decide whether to run in the background (Daemon mode) or foreground if mediamon.options.start is not None: mediamon.start() else: mediamon.run() sys.exit(0) except KeyboardInterrupt: logger.info("Received keyboard interrupt, exiting.") sys.exit(0) except SystemExit, e: """ We just catch this so it won't end up in the catch-all below but we still must raise the exception if we want python to exit with the provided sys.exit() return code """ if not sys.stdin.isatty(): logger.info("SystemExit status %d" % e.code) raise except: exc_type, exc_value, exc_traceback = sys.exc_info() sngpy.print_exception(exc_type, exc_value, exc_traceback) if not sys.stdin.isatty(): logger.error("Unexpected exception %s/%s, aborting." % (exc_type, exc_value)) sys.exit(1)