| #!/usr/bin/python |
| |
| # Copyright (c) 2011 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| # |
| # Implement a pseudo cdma modem. |
| # |
| # This modem mimics a CDMA modem and allows a user to experiment with |
| # a modem which starts in a factory reset state and is gradually moved |
| # into a fully activated state. |
| # |
| # To test you'll need to have a machine with at least 1 ethernet port |
| # available to simulate the cellular connection. Assume that it is |
| # called eth1. |
| # 1. install flimflam-test on your DUT. |
| # 2. sudo backchannel setup eth0 pseudo-modem0 |
| # 3. activation-server & |
| # 4. sudo fake-cromo |
| # 5. Use the UI to "Activate Test Network" |
| # |
| |
| |
| import dbus, glib, gobject, os, subprocess, sys, time |
| from optparse import OptionParser |
| |
| import_path = os.environ.get('SYSROOT', '/usr/local') + '/usr/lib/flimflam/test' |
| sys.path.append(import_path) |
| |
| import flimflam_test |
| |
| Modem = flimflam_test.Modem |
| ModemManager = flimflam_test.ModemManager |
| |
| class IpPoolRestrictor: |
| def __init__(self, interface): |
| self.interface = interface |
| self.restricted = False |
| |
| def enter(self): |
| # Reject all non local tcp traffic, but allow DNS lookups |
| if self.restricted: |
| return |
| subprocess.call(['iptables', |
| '-I', 'INPUT', '1', |
| '-p', 'tcp', |
| '-i', self.interface, |
| '-j', 'REJECT']) |
| self.restricted = True |
| |
| def leave(self, force=None): |
| if self.restricted or force: |
| subprocess.call(['iptables', |
| '-D', 'INPUT', |
| '-p', 'tcp', |
| '-i', self.interface, |
| '-j', 'REJECT']) |
| self.restricted = False |
| |
| restrictor = IpPoolRestrictor('pseudo-modem0') |
| |
| |
| class CarrierState(dbus.service.Object): |
| def __init__(self, bus, path): |
| self.payment_made = False |
| self.restricted = False |
| dbus.service.Object.__init__(self, bus, path) |
| |
| @dbus.service.method('org.chromium.ModemManager.Carrier', |
| in_signature = '', out_signature = '') |
| def ProcessPayment(self, *args, **kwargs): |
| print "CarrierState: ProcessPayment" |
| self.payment_made = True |
| self.restricted = False |
| |
| @dbus.service.method('org.chromium.ModemManager.Carrier', |
| in_signature = '', out_signature = '') |
| def ConsumePlan(self, *args, **kwargs): |
| print "CarrierState: ConsumePlan" |
| self.payment_made = False |
| self.restricted = True |
| |
| |
| class FactoryResetModem(Modem): |
| |
| def __init__(self, mm, name): |
| Modem.__init__(self, mm, name, |
| mdn='0000001234', |
| activation_state=Modem.NOT_ACTIVATED) |
| |
| def Activate(self, s, *args, **kwargs): |
| print 'FactoryResetModem: Activate "%s"' % s |
| self.StartActivation(Modem.PARTIALLY_ACTIVATED, |
| self.manager.MakePartiallyActivatedModem, |
| '0015551212') |
| |
| # Implement connect as a failure |
| def Connect(self, props, *args, **kwargs): |
| print 'FactoryResetModem: Connect' |
| time.sleep(self.manager.options.connect_delay_ms / 1000.0) |
| self.state = flimflam_test.STATE_CONNECTING |
| glib.timeout_add(500, lambda: self.ConnectDone( |
| self.state, |
| flimflam_test.STATE_REGISTERED, |
| flimflam_test.REASON_USER_REQUESTED)) |
| raise flimflam_test.ConnectError() |
| |
| |
| class PartiallyActivatedModem(Modem): |
| |
| def __init__(self, mm, name): |
| Modem.__init__(self, mm, name, |
| mdn='0015551212', |
| activation_state=Modem.PARTIALLY_ACTIVATED) |
| |
| def Activate(self, s, *args, **kwargs): |
| print 'Partially_ActivatedModem: Activate "%s"' % s |
| carrier = self.manager.carrier |
| if self.manager.options.activatable and carrier.payment_made: |
| self.StartActivation(Modem.ACTIVATED, |
| self.manager.MakeActivatedModem, |
| '6175551212') |
| else: |
| # TODO(jglasgow): define carrier error codes |
| carrier_error = 1 |
| self.StartFailedActivation(carrier_error) |
| |
| def ConnectDone(self, old, new, why): |
| # Implement ConnectDone by manipulating the IP pool restrictor |
| if new == flimflam_test.STATE_CONNECTED: |
| restrictor.enter() |
| else: |
| restrictor.leave() |
| super(PartiallyActivatedModem, self).ConnectDone(old, new, why) |
| |
| |
| class ActivatedModem(Modem): |
| def __init__(self, mm, name): |
| Modem.__init__(self, mm, name, |
| mdn='6175551212', |
| activation_state=Modem.ACTIVATED) |
| |
| def ConnectDone(self, old, new, why): |
| carrier = self.manager.carrier |
| # Implement ConnectDone by manipulating the IP pool restrictor |
| if new == flimflam_test.STATE_CONNECTED and carrier.restricted: |
| restrictor.enter() |
| else: |
| restrictor.leave() |
| super(ActivatedModem, self).ConnectDone(old, new, why) |
| |
| def Connect(self, props, *args, **kwargs): |
| print 'ActivatedModem: Connect' |
| kwargs['connect_delay_ms'] = ( |
| self.manager.options.connect_delay_ms) |
| Modem.Connect(self, props, *args, **kwargs) |
| |
| |
| class BrokenActivatedModem(Modem): |
| """ BrokenActivatedModem is a modem that although activated always |
| fails to connect to the network. This simulates errors in which the |
| carrier refuses to allow connections. |
| """ |
| def __init__(self, mm, name): |
| Modem.__init__(self, mm, name, |
| mdn='6175551212', |
| activation_state=Modem.ACTIVATED) |
| |
| # Implement connect by always failing |
| def Connect(self, props, *args, **kwargs): |
| print 'BrokenActivatedModem: Connect' |
| time.sleep(self.manager.options.connect_delay_ms / 1000.0) |
| self.state = flimflam_test.STATE_CONNECTING |
| glib.timeout_add(500, lambda: self.ConnectDone( |
| self.state, |
| flimflam_test.STATE_REGISTERED, |
| flimflam_test.REASON_USER_REQUESTED)) |
| raise flimflam_test.ConnectError() |
| |
| |
| class Manager(ModemManager): |
| def __init__(self, bus, options): |
| ModemManager.__init__(self, bus, flimflam_test.OCMM) |
| self.modem_number = 1 |
| self.options = options |
| self.carrier = CarrierState(bus, |
| '/org/chromium/ModemManager/Carrier') |
| |
| def NewModem(self, classname): |
| # modem registeres itself with mm, so does not disappear |
| _ = classname(self, '/TestModem/%d' % self.modem_number) |
| self.modem_number += 1 |
| |
| def MakeFactoryResetModem(self): |
| self.NewModem(FactoryResetModem) |
| |
| def MakePartiallyActivatedModem(self): |
| self.NewModem(PartiallyActivatedModem) |
| |
| def MakeActivatedModem(self): |
| if not self.options.activatable: |
| self.NewModem(PartiallyActivatedModem) |
| elif self.options.connectable: |
| self.NewModem(ActivatedModem) |
| else: |
| self.NewModem(BrokenActivatedModem) |
| |
| |
| def main(): |
| usage = ''' |
| Run the fake cromo program to simulate different modem and carrier |
| behaviors. By default with no arguments the program will simulate a |
| factory reset modem which needs to be activated and requires the user |
| to sign up for service. |
| |
| To test for error cases in which connections to the carrier network |
| always fail, use the --unconnectable flag. This is particularly |
| useful when the initial state is set to activated as in: |
| |
| sudo fake-cromo -u -s activated |
| |
| This can be used to simulate the conditions of crosbug.com/11355 |
| |
| Another simulation that corresponds to many field error conditions is |
| a device that fails OTASP activation. This can be simulated by using |
| the -a flag. The device should start in either the factory or partial |
| state. |
| |
| sudo fake-cromo -a |
| |
| To test the re-up process, start the modem in the restricted state with |
| |
| sudo fake-cromo -s restricted |
| |
| One can leave the restricted state by fetching |
| http://localhost:8080/payment_succeeded.html. This can be done via |
| the UI by pressing "Buy Plan", or manually. On the next reconnect the |
| user should be out of the restricted IP pool. |
| |
| If the program is interupted while a partially activated modem is in |
| the connected state it may leave the iptables set up in a way that |
| causes all tcp traffic to fail, even when using other network |
| interfaces. Fix this by restarting fake cromo and tell it to leave |
| the restricted IP pool |
| |
| sudo fake-cromo -l |
| |
| ''' |
| parser = OptionParser(usage=usage) |
| parser.add_option('-u', '--unconnectable', |
| action='store_false', dest='connectable', |
| default=True, |
| help='Do not allow modem to connect') |
| parser.add_option('-s', '--state', dest='initial_state', |
| type='choice', |
| choices=['factory', 'partial', |
| 'activated', 'restricted'], |
| default='factory', |
| help=('Set initial state to factory,' |
| 'partial, restricted or activated')) |
| parser.add_option('-a', '--activation_fails', |
| action='store_false', dest='activatable', |
| default=True, |
| help='Do not allow modem to activate') |
| parser.add_option('-l', '--leave-restricted-pool', |
| action='store_true', dest='leave_restricted_pool', |
| default=False, |
| help='Leave the restricted pool and exit') |
| parser.add_option('--connect-delay', type='int', |
| dest='connect_delay_ms', |
| default=flimflam_test.DEFAULT_CONNECT_DELAY_MS, |
| help='time in ms required to connnect') |
| |
| (options, args) = parser.parse_args() |
| if len(args) != 0: |
| parser.error("incorrect number of arguments") |
| |
| dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) |
| bus = dbus.SystemBus() |
| mm = Manager(bus, options) |
| mainloop = gobject.MainLoop() |
| print "Running test modemmanager." |
| _ = dbus.service.BusName(flimflam_test.CMM, bus) |
| |
| if options.leave_restricted_pool: |
| restrictor.leave(force=True) |
| return 0 |
| |
| # Choose the type of modem to instantiate... |
| if options.initial_state == 'factory': |
| mm.MakeFactoryResetModem() |
| elif options.initial_state == 'partial': |
| mm.MakePartiallyActivatedModem() |
| elif options.initial_state == 'activated': |
| mm.MakeActivatedModem() |
| elif options.initial_state == 'restricted': |
| mm.carrier.ConsumePlan() |
| mm.MakeActivatedModem() |
| else: |
| print 'Invalid initial state: %s' % options.initial_state |
| return 1 |
| |
| try: |
| mainloop.run() |
| finally: |
| restrictor.leave(force=True) |
| |
| |
| if __name__ == '__main__': |
| main() |
