blob: 1ab61aee86c270edb2864eaf8a4d6ceaa14612ff [file] [log] [blame]
#!/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()
def ActivateImpl(self, _s, _args, _kwargs):
raise NotImplementedError('Unimplemented. Must implement in subclass.')
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()
Modem.ConnectDone(self, old, new, why)
def ActivateImpl(self, _s, _args, _kwargs):
raise NotImplementedError('Unimplemented. Must implement in subclass.')
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()
Modem.ConnectDone(self, 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)
def ActivateImpl(self, _s, _args, _kwargs):
raise NotImplementedError('Unimplemented. Must implement in subclass.')
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()
def ActivateImpl(self, _s, _args, _kwargs):
raise NotImplementedError('Unimplemented. Must implement in subclass.')
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()