colin/ModemManager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: add ability to drive ppp for connection tests

Browse Source
This commit is contained in:
Dan Williams
2009-12-20 00:17:12 -08:00
parent 87ee623923
commit a8fc7ecd77
4 changed files with 432 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
configure.ac
View File

@@ -39,6 +39,13 @@ AC_SUBST(UDEV_BASE_DIR)
GLIB_GENMARSHAL=`pkg-config --variable=glib_genmarshal glib-2.0`
AC_SUBST(GLIB_GENMARSHAL)
if test -n "$with_pppd_plugin_dir" ; then
PPPD_PLUGIN_DIR="$with_pppd_plugin_dir"
else
PPPD_PLUGIN_DIR="${libdir}/pppd/2.4.4"
fi
AC_SUBST(PPPD_PLUGIN_DIR)
AC_ARG_ENABLE(more-warnings,
AS_HELP_STRING([--enable-more-warnings], [Maximum compiler warnings]), set_more_warnings="$enableval",set_more_warnings=yes)
AC_MSG_CHECKING(for more warnings, including -Werror)

16
test/Makefile.am
View File

@@ -1,3 +1,19 @@
pppd_plugindir = $(PPPD_PLUGIN_DIR)
pppd_plugin_LTLIBRARIES = mm-test-pppd-plugin.la
mm_test_pppd_plugin_la_SOURCES = \
mm-test-pppd-plugin.c
mm_test_pppd_plugin_la_CPPFLAGS = $(MM_CFLAGS)
mm_test_pppd_plugin_la_LDFLAGS = -module -avoid-version
mm_test_pppd_plugin_la_LIBADD = $(MM_LIBS)
noinst_PROGRAMS = lsudev
lsudev_SOURCES = lsudev.c
lsudev_CPPFLAGS = $(GUDEV_CFLAGS)
lsudev_LDADD = $(GUDEV_LIBS)
EXTRA_DIST = \
mm-test.py

264
test/mm-test-pppd-plugin.c Normal file
View File

@@ -0,0 +1,264 @@
/* -*- Mode: C; tab-width: 4; indent-tabs-mode: t; c-basic-offset: 4 -*- */
/* 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.
*
* Copyright (C) 2008 Novell, Inc.
* Copyright (C) 2008 - 2009 Red Hat, Inc.
*/
#include <string.h>
#include <pppd/pppd.h>
#include <pppd/fsm.h>
#include <pppd/ipcp.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <glib.h>
int plugin_init (void);
char pppd_version[] = VERSION;
char *my_user = NULL;
char *my_pass = NULL;
char *my_file = NULL;
static void
mm_phasechange (void *data, int arg)
{
const char *ppp_phase = NULL;
switch (arg) {
case PHASE_DEAD:
ppp_phase = "dead";
break;
case PHASE_INITIALIZE:
ppp_phase = "initialize";
break;
case PHASE_SERIALCONN:
ppp_phase = "serial connection";
break;
case PHASE_DORMANT:
ppp_phase = "dormant";
break;
case PHASE_ESTABLISH:
ppp_phase = "establish";
break;
case PHASE_AUTHENTICATE:
ppp_phase = "authenticate";
break;
case PHASE_CALLBACK:
ppp_phase = "callback";
break;
case PHASE_NETWORK:
ppp_phase = "network";
break;
case PHASE_RUNNING:
ppp_phase = "running";
break;
case PHASE_TERMINATE:
ppp_phase = "terminate";
break;
case PHASE_DISCONNECT:
ppp_phase = "disconnect";
break;
case PHASE_HOLDOFF:
ppp_phase = "holdoff";
break;
case PHASE_MASTER:
ppp_phase = "master";
break;
default:
ppp_phase = "unknown";
break;
}
g_message ("mm-test-ppp-plugin: (%s): phase now '%s'", __func__, ppp_phase);
}
static void
append_ip4_addr (GString *str, const char *tag, guint32 addr)
{
char buf[INET_ADDRSTRLEN + 2];
struct in_addr tmp_addr = { .s_addr = addr };
memset (buf, 0, sizeof (buf));
if (inet_ntop (AF_INET, &tmp_addr, buf, sizeof (buf) - 1))
g_string_append_printf (str, "%s %s\n", tag, buf);
}
static void
mm_ip_up (void *data, int arg)
{
ipcp_options opts = ipcp_gotoptions[0];
ipcp_options peer_opts = ipcp_hisoptions[0];
guint32 pppd_made_up_address = htonl (0x0a404040 + ifunit);
GString *contents;
GError *err = NULL;
gboolean success;
g_message ("mm-test-ppp-plugin: (%s): ip-up event", __func__);
if (!opts.ouraddr) {
g_warning ("mm-test-ppp-plugin: (%s): didn't receive an internal IP from pppd!", __func__);
mm_phasechange (NULL, PHASE_DEAD);
return;
}
contents = g_string_sized_new (100);
g_string_append_printf (contents, "iface %s\n", ifname);
append_ip4_addr (contents, "addr", opts.ouraddr);
/* Prefer the peer options remote address first, _unless_ pppd made the
* address up, at which point prefer the local options remote address,
* and if that's not right, use the made-up address as a last resort.
*/
if (peer_opts.hisaddr && (peer_opts.hisaddr != pppd_made_up_address))
append_ip4_addr (contents, "gateway", peer_opts.hisaddr);
else if (opts.hisaddr)
append_ip4_addr (contents, "gateway", opts.hisaddr);
else if (peer_opts.hisaddr == pppd_made_up_address) {
/* As a last resort, use the made-up address */
append_ip4_addr (contents, "gateway", peer_opts.hisaddr);
}
if (opts.dnsaddr[0] || opts.dnsaddr[1]) {
if (opts.dnsaddr[0])
append_ip4_addr (contents, "dns1", opts.dnsaddr[0]);
if (opts.dnsaddr[1])
append_ip4_addr (contents, "dns2", opts.dnsaddr[1]);
}
if (opts.winsaddr[0] || opts.winsaddr[1]) {
if (opts.winsaddr[0])
append_ip4_addr (contents, "wins1", opts.winsaddr[0]);
if (opts.winsaddr[1])
append_ip4_addr (contents, "wins2", opts.winsaddr[1]);
}
g_string_append (contents, "DONE\n");
success = g_file_set_contents (my_file, contents->str, -1, &err);
if (success)
g_message ("nm-ppp-plugin: (%s): saved IP4 config to %s", __func__, my_file);
else {
g_message ("nm-ppp-plugin: (%s): error saving IP4 config to %s: (%d) %s",
__func__, my_file, err->code, err->message);
g_clear_error (&err);
}
g_string_free (contents, TRUE);
}
static int
get_chap_check()
{
return 1;
}
static int
get_pap_check()
{
return 1;
}
static int
get_credentials (char *username, char *password)
{
size_t len;
if (username && !password) {
/* pppd is checking pap support; return 1 for supported */
return 1;
}
g_message ("nm-ppp-plugin: (%s): sending credentials (%s / %s)",
__func__,
my_user ? my_user : "",
my_pass ? my_pass : "");
if (my_user) {
len = strlen (my_user) + 1;
len = len < MAXNAMELEN ? len : MAXNAMELEN;
strncpy (username, my_user, len);
username[len - 1] = '\0';
}
if (my_pass) {
len = strlen (my_pass) + 1;
len = len < MAXSECRETLEN ? len : MAXSECRETLEN;
strncpy (password, my_pass, len);
password[len - 1] = '\0';
}
return 1;
}
static void
mm_exit_notify (void *data, int arg)
{
g_message ("mm-test-ppp-plugin: (%s): cleaning up", __func__);
g_free (my_user);
my_user = NULL;
g_free (my_pass);
my_pass = NULL;
g_free (my_file);
my_file = NULL;
}
int
plugin_init (void)
{
char **args;
g_message ("mm-test-ppp-plugin: (%s): initializing", __func__);
/* mm-test passes the file + username + password in as the 'ipparam' arg
* to pppd.
*/
args = g_strsplit (ipparam, "+", 0);
if (!args || g_strv_length (args) != 3) {
g_message ("mm-test-ppp-plugin: (%s): ipparam arguments error ('%s')",
__func__, ipparam);
return -1;
}
my_user = (args[0] && strlen (args[0])) ? g_strdup (args[0]) : NULL;
my_pass = (args[1] && strlen (args[1])) ? g_strdup (args[1]) : NULL;
my_file = (args[2] && strlen (args[2])) ? g_strdup (args[2]) : NULL;
g_strfreev (args);
if (!my_file) {
g_message ("mm-test-ppp-plugin: (%s): missing IP config file",
__func__);
return -1;
}
chap_passwd_hook = get_credentials;
chap_check_hook = get_chap_check;
pap_passwd_hook = get_credentials;
pap_check_hook = get_pap_check;
add_notifier (&phasechange, mm_phasechange, NULL);
add_notifier (&ip_up_notifier, mm_ip_up, NULL);
add_notifier (&exitnotify, mm_exit_notify, NULL);
return 0;
}

154
test/mm-test.py
View File

@@ -15,9 +15,7 @@
# Copyright (C) 2009 Red Hat, Inc.
#
import sys
import dbus
import time
import sys, dbus, time, os, string, subprocess
DBUS_INTERFACE_PROPERTIES='org.freedesktop.DBus.Properties'
MM_DBUS_SERVICE='org.freedesktop.ModemManager'
@@ -87,8 +85,10 @@ def cdma_connect(proxy, user, password):
try:
simple.Connect({'number':"#777"}, timeout=60)
print "\nConnected!"
return True
except Exception, e:
print "Error connecting: %s" % e
return False
def get_gsm_network_mode(modem):
@@ -156,7 +156,7 @@ def get_gsm_band(modem):
print "Band: %s" % band
def gsm_inspect(proxy, dump_private):
def gsm_inspect(proxy, dump_private, do_scan):
# Gsm.Card interface
card = dbus.Interface(proxy, dbus_interface=MM_DBUS_INTERFACE_MODEM_GSM_CARD)
@@ -183,6 +183,9 @@ def gsm_inspect(proxy, dump_private):
except dbus.exceptions.DBusException, e:
print "Error reading signal quality: %s" % e
if not do_scan:
return
print "Scanning..."
try:
results = net.Scan(timeout=120)
@@ -239,8 +242,109 @@ def gsm_connect(proxy, apn, user, password):
opts['password'] = password
simple.Connect(opts, timeout=120)
print "\nConnected!"
return True
except Exception, e:
print "Error connecting: %s" % e
return False
def pppd_find():
paths = ["/usr/local/sbin/pppd", "/usr/sbin/pppd", "/sbin/pppd"]
for p in paths:
if os.path.exists(p):
return p
return None
def ppp_start(device, user, password, tmpfile):
path = pppd_find()
if not path:
return None
args = [path]
args += ["nodetach"]
args += ["lock"]
args += ["nodefaultroute"]
args += ["debug"]
if user:
args += ["user"]
args += [user]
args += ["noipdefault"]
args += ["115200"]
args += ["noauth"]
args += ["crtscts"]
args += ["modem"]
args += ["usepeerdns"]
args += ["ipparam"]
ipparam = ""
if user:
ipparam += user
ipparam += "+"
if password:
ipparam += password
ipparam += "+"
ipparam += tmpfile
args += [ipparam]
args += ["plugin"]
args += ["mm-test-pppd-plugin.so"]
args += [device]
return subprocess.Popen(args, close_fds=True, cwd="/", env={})
def ppp_wait(p, tmpfile):
i = 0
while p.poll() == None and i < 30:
time.sleep(1)
if os.path.exists(tmpfile):
f = open(tmpfile, 'r')
stuff = f.read(500)
idx = string.find(stuff, "DONE")
f.close()
if idx >= 0:
return True
i += 1
return False
def configure_iface(tmpfile):
addr = None
gw = None
iface = None
dns1 = None
dns2 = None
f = open(tmpfile, 'r')
lines = f.readlines()
for l in lines:
if l.startswith("addr"):
addr = l[len("addr"):].strip()
if l.startswith("gateway"):
gw = l[len("gateway"):].strip()
if l.startswith("iface"):
iface = l[len("iface"):].strip()
if l.startswith("dns1"):
dns1 = l[len("dns1"):].strip()
if l.startswith("dns2"):
dns2 = l[len("dns2"):].strip()
f.close()
print "\n******************************"
print "iface: %s" % iface
print "addr: %s" % addr
print "gw: %s" % gw
print "dns1: %s" % dns1
print "dns2: %s" % dns2
ifconfig = ["ifconfig", iface, "%s/32" % addr, "dstaddr", gw]
print " ".join(ifconfig)
print "\n******************************"
subprocess.call(ifconfig)
def ppp_stop(p):
import signal
p.send_signal(signal.SIGTERM)
p.wait()
dump_private = False
@@ -248,6 +352,8 @@ connect = False
apn = None
user = None
password = None
do_ip = False
do_scan = True
x = 1
while x < len(sys.argv):
if sys.argv[x] == "--private":
@@ -263,6 +369,10 @@ while x < len(sys.argv):
elif sys.argv[x] == "--password":
x += 1
password = sys.argv[x]
elif sys.argv[x] == "--ip":
do_ip = True
elif sys.argv[x] == "--no-scan":
do_scan = False
x += 1
bus = dbus.SystemBus()
@@ -277,6 +387,9 @@ if not modems:
sys.exit(1)
for m in modems:
connect_success = False
data_device = None
proxy = bus.get_object(MM_DBUS_SERVICE, m)
# Properties
@@ -292,11 +405,17 @@ for m in modems:
print "Driver: '%s'" % (props_iface.Get(MM_DBUS_INTERFACE_MODEM, 'Driver'))
print "Modem device: '%s'" % (props_iface.Get(MM_DBUS_INTERFACE_MODEM, 'MasterDevice'))
print "Data device: '%s'" % (props_iface.Get(MM_DBUS_INTERFACE_MODEM, 'Device'))
data_device = props_iface.Get(MM_DBUS_INTERFACE_MODEM, 'Device')
print "Data device: '%s'" % data_device
# Modem interface
modem = dbus.Interface(proxy, dbus_interface=MM_DBUS_INTERFACE_MODEM)
modem.Enable(True)
try:
modem.Enable(True)
except dbus.exceptions.DBusException, e:
print "Error enabling modem: %s" % e
sys.exit(1)
info = modem.GetInfo()
print "Vendor: %s" % info[0]
@@ -304,15 +423,32 @@ for m in modems:
print "Version: %s" % info[2]
if type == 1:
gsm_inspect(proxy, dump_private)
gsm_inspect(proxy, dump_private, do_scan)
if connect == True:
gsm_connect(proxy, apn, user, password)
connect_success = gsm_connect(proxy, apn, user, password)
elif type == 2:
cdma_inspect(proxy, dump_private)
if connect == True:
cdma_connect(proxy, user, password)
connect_success = cdma_connect(proxy, user, password)
print
if connect_success and do_ip:
tmpfile = "/tmp/mm-test-%d.tmp" % os.getpid()
try:
p = ppp_start(data_device, user, password, tmpfile)
if ppp_wait(p, tmpfile):
configure_iface(tmpfile)
time.sleep(30)
ppp_stop(p)
modem.Disconnect()
except Exception, e:
print "Error handling PPP: %s" % e
try:
os.remove(tmpfile)
except:
pass
time.sleep(5)
modem.Enable(False)