blob: 08b44d257020911eb0af2667fc1ce6b941ba0126 [file] [log] [blame]
# Copyright (c) 2013 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.
# This is an integration test which ensures that a proxy set on a
# shared network connection is exposed via LibCrosSevice and used
# by tlsdated during time synchronization.
import dbus
import gobject
import logging
import subprocess
import threading
import time
from autotest_lib.client.bin import test, utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.cros import cros_ui
from autotest_lib.client.cros.networking import shill_proxy
from dbus.mainloop.glib import DBusGMainLoop
from SocketServer import ThreadingTCPServer, StreamRequestHandler
class ProxyHandler(StreamRequestHandler):
"""Matching request handler for the ThreadedHitServer
that notes when an expected request is seen.
wbufsize = -1
def handle(self):
"""Reads the first line, up to 40 characters, looking
for the CONNECT string that tlsdated sends. If it
is found, the server's hit() method is called.
All requests receive a HTTP 504 error.
# Read up to 40 characters
data = self.rfile.readline(40).strip()'ProxyHandler::handle(): <%s>', data)
# TODO(wad) Add User-agent check when it lands in tlsdate.
# Also, abstract the time server and move this code into cros/.
if data.__contains__('CONNECT HTTP/1.1'):
self.wfile.write("HTTP/1.1 504 Gateway Timeout\r\n" +
"Connection: close\r\n\r\n")
class ThreadedHitServer(ThreadingTCPServer):
"""A threaded TCP server which services requests
and allows the handler to track "hits".
def __init__(self, server_address, HandlerClass):
@param server_address: tuple of server IP and port to listen on.
@param HandlerClass: the RequestHandler class to instantiate per req.
self._hits = 0
ThreadingTCPServer.__init__(self, server_address, HandlerClass)
def hit(self):
"""Increment the hit count. Usually called by the HandlerClass"""
self._hits += 1
def reset_hits(self):
"""Set the hit count to 0"""
self._hits = 0
def hits(self):
"""Get the number of matched requests
@return the count of matched requests
return self._hits
class ProxyListener(object):
"""A fake listener for tracking if an expected CONNECT request is
seen at the provided server address. Any hits are exposed to be
consumed by the caller.
def __init__(self, server_address):
@param server_address: tuple of server IP and port to listen on.
self._server = ThreadedHitServer(server_address, ProxyHandler)
self._thread = threading.Thread(target=self._server.serve_forever)
def run(self):
"""Run the server on a thread"""
def stop(self):
"""Stop the server and its threads"""
def reset_hits(self):
"""Reset the number of matched requests to 0"""
return self._server.reset_hits()
def hits(self):
"""Get the number of matched requests
@return the count of matched requests
return self._server.hits()
class SignalListener(object):
"""A class to listen for a DBus signal
_main_loop = None
_signals = { }
def __init__(self, g_main_loop):
@param g_mail_loop: glib main loop object.
self._main_loop = g_main_loop
def listen_for_signal(self, signal, interface, path):
"""Listen with a default handler
@param signal: signal name to listen for
@param interface: DBus interface to expect it from
@param path: DBus path associated with the signal
self.__listen_to_signal(self.__handle_signal, signal, interface, path)
def wait_for_signals(self, desc,
"""Block for |timeout| seconds waiting for the signals to come in.
@param desc: string describing the high-level reason you're waiting
for the signals.
@param timeout: maximum seconds to wait for the signals.
@raises TimeoutError if the timeout is hit.
condition=lambda: self.__received_signals(),
all_signals = self._signals.copy()
return all_signals
def __received_signals(self):
"""Run main loop until all pending events are done, checks for signals.
Runs self._main_loop until it says it has no more events pending,
then returns the state of the internal variables tracking whether
desired signals have been received.
@return True if both signals have been handled, False otherwise.
context = self._main_loop.get_context()
while context.iteration(False):
return len(self._signals) > 0
def __reset_signal_state(self):
"""Resets internal signal tracking state."""
self._signals = { }
def __listen_to_signal(self, callback, signal, interface, path):
"""Connect a callback to a given session_manager dbus signal.
Sets up a signal receiver for signal, and calls the provided callback
when it comes in.
@param callback: a callable to call when signal is received.
@param signal: the signal to listen for.
bus = dbus.SystemBus(mainloop=self._main_loop)
def __handle_signal(self, *args, **kwargs):
"""Callback to be used when a new key signal is received."""
signal_name = kwargs.pop('signal_name', '')
#signal_data = str(args[0])"SIGNAL: " + signal_name + ", " + str(args));
if self._signals.has_key(signal_name):
self._signals[signal_name] = [args]
class network_ProxyResolver(test.test):
"""A test fixture for validating the integration of
shill, Chrome, and tlsdated's proxy resolution.
version = 1
auto_login = False
service_settings = { }
def initialize(self):
Sets up the test such that all DBus signals can be
received and a fake proxy server can be instantiated.
Additionally, the UI is restarted to ensure consistent
shared network use.
super(network_ProxyResolver, self).initialize()
self._listener = SignalListener(gobject.MainLoop())
self._shill = shill_proxy.ShillProxy.get_proxy()
if self._shill is None:
raise error.TestFail('Could not connect to shill')
# Listen for network property changes
# Listen on the proxy port.
self._proxy_server = ProxyListener(('', 3128))
# Set the proxy with Shill. This only works for shared connections
# (like Eth).
def set_proxy(self, service_name, proxy_config):
"""Changes the ProxyConfig property on the specified shill service.
@param service_name: the name, as a str, of the shill service
@param proxy_config: the ProxyConfig property value string
@raises TestFail if the service is not found.
shill = self._shill
service = shill.find_object('Service', { 'Name' : service_name })
if not service:
raise error.TestFail('Service ' + service_name +
' not found to test proxy with.')
props = service.GetProperties()
old_proxy = ''
if props.has_key('ProxyConfig'):
old_proxy = props['ProxyConfig']
if self.service_settings.has_key(service_name) == False:'Preexisting ProxyConfig: ' + service_name +
' -> ' + old_proxy)
self.service_settings[service_name] = old_proxy'Setting proxy to ' + proxy_config)
service.SetProperties({'ProxyConfig': proxy_config})
def reset_services(self):
"""Walks the dict of service->ProxyConfig values and sets the
proxy back to the originally observed value.
if len(self.service_settings) == 0:
for k,v in self.service_settings.items():'Resetting ProxyConfig: ' + k + ' -> ' + v)
self.set_proxy(k, v)
def check_chrome(self, proxy_type, proxy_config, timeout):
"""Check that Chrome has acknowledged the supplied proxy config
by asking for resolution over DBus.
@param proxy_type: PAC-style string type (e.g., 'PROXY', 'SOCKS')
@param proxy_config: PAC-style config string (e.g.,
@param timeout: time in seconds to wait for Chrome to issue a signal.
@return True if a matching response is seen and False otherwise
bus = dbus.SystemBus()
dbus_proxy = bus.get_object('org.chromium.NetworkProxyService',
service = dbus.Interface(dbus_proxy,
attempts = timeout
while attempts > 0:
result, _ = service.ResolveProxy('')
if str(result) == proxy_type + ' ' + proxy_config:
return True
attempts -= 1
logging.error('Last response seen before giving up: ' + str(result))
return False
def check_tlsdated(self, timeout):
"""Check that tlsdated uses the set proxy.
@param timeout: time in seconds to wait for tlsdate to restart and query
@return True if tlsdated hits the proxy server and False otherwise
# Restart tlsdated to force a network resync
# (The other option is to force it to think there is no network sync.)
except Exception as e:
logging.error("Proxy error =>" + str(e))
return False"proxy started!")
status =['initctl', 'restart', 'tlsdated'])
if status != 0:"failed to restart tlsdated")
return False
attempts = timeout"waiting for hits on the proxy server")
while attempts > 0:
if self._proxy_server.hits() > 0:
return True
attempts -= 1"no hits")
return False
def cleanup(self):
"""Reset all the service data and teardown the proxy."""
self.reset_services()"tearing down the proxy server")
self._proxy_server.stop()"proxy server down")
super(network_ProxyResolver, self).cleanup()
def test_same_ip_proxy_at_signin_chrome_system_tlsdated(
""" Set the user policy, waits for condition, then logs out.
@param service_name: shill service name to test on
@param test_timeout: the total time in seconds split among all timeouts.
proxy_type = 'http'
proxy_port = '3128'
proxy_host = ''
proxy_url = proxy_type + '://' + proxy_host + ':' + proxy_port
# TODO(wad) Only do the below if it was a single protocol proxy.
# proxy_config = proxy_type + '=' + proxy_host + ':' + proxy_port
proxy_config = proxy_host + ':' + proxy_port
self.set_proxy(service_name, '{"mode":"fixed_servers","server":"' +
proxy_config + '"}')"checking chrome")
if self.check_chrome('PROXY', proxy_config, test_timeout/3) == False:
raise error.TestFail('Chrome failed to resolve the proxy')
# Restart tlsdate to force a network fix"checking tlsdated")
if self.check_tlsdated(test_timeout/3) == False:
raise error.TestFail('tlsdated never tried the proxy')"done!")
def run_once(self, test_type, **params):'client: Running client test %s', test_type)
getattr(self, test_type)(**params)