blob: b0ba529bfd2d99e1f1baf9a03eaae98fb041c59b [file] [log] [blame]
#!/usr/bin/env python
# Copyright (c) 2012 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Helper script to perform actions as a super-user on ChromeOS.
Needs to be run with superuser privileges, typically using the
suid_python binary.
sudo python --action=CleanFlimflamDirs
import optparse
import os
import shutil
import subprocess
import sys
import time
sys.path.append('/usr/local') # to import autotest libs.
from autotest.cros import constants
from autotest.cros import cryptohome
TEMP_BACKCHANNEL_FILE = '/tmp/pyauto_network_backchannel_file'
class SuidAction(object):
"""Helper to perform some super-user actions on ChromeOS."""
def _ParseArgs(self):
parser = optparse.OptionParser()
'-a', '--action', help='Action to perform.')
self._options = parser.parse_args()[0]
if not self._options.action:
raise RuntimeError('No action specified.')
def Run(self):
assert os.geteuid() == 0, 'Needs superuser privileges.'
handler = getattr(self, self._options.action)
assert handler and callable(handler), \
'No handler for %s' % self._options.action
return 0
## Actions ##
def CleanFlimflamDirs(self):
"""Clean the contents of all connection manager (shill/flimflam) profiles.
flimflam_dirs = ['/home/chronos/user/flimflam',
# The stop/start flimflam command should stop/start shill respectivly if
# enabled.
os.system('stop flimflam')
for flimflam_dir in flimflam_dirs:
if not os.path.exists(flimflam_dir):
for item in os.listdir(flimflam_dir):
path = os.path.join(flimflam_dir, item)
if os.path.isdir(path):
os.system('start flimflam')
# TODO(stanleyw): This method should wait until
# flimflam/shill is fully initialized and accessible via DBus again.
# Otherwise, there is a race conditions and subsequent accesses to
# flimflam/shill may fail. Until this is fixed, waiting for the
# resolv.conf file to be created is better than nothing.
begin = time.time()
while not os.path.exists(constants.RESOLV_CONF_FILE):
if time.time() - begin > 10:
raise RuntimeError('Timeout while waiting for flimflam/shill start.')
def RemoveAllCryptohomeVaults(self):
"""Remove any existing cryptohome vaults."""
def _GetEthInterfaces(self):
"""Returns a list of the eth* interfaces detected by the device."""
# Assumes ethernet interfaces all have "eth" in the name.
import pyudev
return sorted([iface.sys_name for iface in
if 'eth' in iface.sys_name])
def _Renameif(self, old_iface, new_iface, mac_address):
"""Renames the interface with mac_address from old_iface to new_iface.
old_iface: The name of the interface you want to change.
new_iface: The name of the interface you want to change to.
mac_address: The mac address of the interface being changed.
"""['stop', 'flimflam'])['ifconfig', old_iface, 'down'])['nameif', new_iface, mac_address])['ifconfig', new_iface, 'up'])['start', 'flimflam'])
# Check and make sure interfaces have been renamed
eth_ifaces = self._GetEthInterfaces()
if new_iface not in eth_ifaces:
raise RuntimeError('Interface %s was not renamed to %s' %
(old_iface, new_iface))
elif old_iface in eth_ifaces:
raise RuntimeError('Old iface %s is still present' % old_iface)
def SetupBackchannel(self):
"""Renames the connected ethernet interface to eth_test for offline mode
testing. Does nothing if no connected interface is found.
# Return the interface with ethernet connected or returns if none found.
for iface in self._GetEthInterfaces():
with open('/sys/class/net/%s/operstate' % iface, 'r') as fp:
if 'up' in
eth_iface = iface
# Write backup file to be used by TeardownBackchannel to restore the
# interface names.
with open(TEMP_BACKCHANNEL_FILE, 'w') as fpw:
with open('/sys/class/net/%s/address' % eth_iface) as fp:
mac_address =
fpw.write('%s, %s' % (eth_iface, mac_address))
self._Renameif(eth_iface, 'eth_test', mac_address)
def TeardownBackchannel(self):
"""Restores the eth interface names if SetupBackchannel was called."""
if not os.path.isfile(TEMP_BACKCHANNEL_FILE):
with open(TEMP_BACKCHANNEL_FILE, 'r') as fp:
eth_iface, mac_address =',')
self._Renameif('eth_test', eth_iface, mac_address)
if __name__ == '__main__':