Run 2to3

[oddments/fireinfo.git] / src / fireinfo / system.py

#!/usr/bin/python
###############################################################################
#                                                                             #
# Fireinfo                                                                    #
# Copyright (C) 2010, 2011 IPFire Team (www.ipfire.org)                       #
#                                                                             #
# 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 3 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, see <http://www.gnu.org/licenses/>.       #
#                                                                             #
###############################################################################

import hashlib
import json
import os
import string

import _fireinfo

from . import bios
from . import cpu
from . import device
from . import hypervisor
from . import network

PROFILE_VERSION = 0

SYS_CLASS_DMI = "/sys/class/dmi/id"
SECRET_ID_FILE = "/etc/fireinfo-id"

INVALID_ID_STRINGS = (
	"OEM", "O.E.M.", "o.e.m.",
	"N/A", "n/a",
	"12345", "54321", "202020",
	"Chassis", "chassis",
	"Default string",
	"EVAL",
	"Not Applicable",
	"None", "empty",
	"Serial", "System Serial Number",
	"XXXXX",
	"01010101-0101-0101-0101-010101010101",
	"00020003-0004-0005-0006-000700080009",
	"03000200-0400-0500-0006-000700080009",
	"11111111-1111-1111-1111-111111111111",
	"0000000", "00000000",
)

INVALID_ID_STRINGS_EXACT_MATCH = (
	"NA",
)

class Singleton(type):
	def __init__(cls, name, bases, dict):
		super(Singleton, cls).__init__(name, bases, dict)
		cls.instance = None

	def __call__(cls, *args, **kw):
		if cls.instance is None:
			cls.instance = super(Singleton, cls).__call__(*args, **kw)

		return cls.instance


def read_from_file(filename):
	"""
		Read all data from filename.
	"""
	if not os.path.exists(filename):
		return

	try:
		with open(filename) as f:
			return f.read().strip()
	except IOError:
		pass

class System(object, metaclass=Singleton):
	def __init__(self):
		self.bios = bios.BIOS(self)

		# find all devices
		self.devices = []
		self.scan()
		self.cpu = cpu.CPU()
		self.hypervisor = hypervisor.Hypervisor()

		# Read /proc/cpuinfo for vendor information.
		self.__cpuinfo = self.cpu.read_cpuinfo()

	def profile(self):
		p = {}
		p["system"] = {
			# System information
			"model"  : self.model,
			"vendor" : self.vendor,

			# Indicator if the system is running in a
			# virtual environment.
			"virtual" : self.virtual,
			
			# System language
			"language" : self.language,

			# Release information
			"release" : self.release,
			"kernel_release" : self.kernel_release,

			"memory" : self.memory,
			"root_size" : self.root_size,
		}

		p["devices"] = []
		for device in self.devices:
			d = {
				"subsystem" : device.subsystem.lower(), 
				"vendor" : device.vendor.lower(), 
				"model" : device.model.lower(), 
				"deviceclass" : device.deviceclass,
				"driver" : device.driver,
			}

			# PCI devices provide subsystem information, USB don't.
			if d["subsystem"] == "pci":
				d["sub_model"] = device.sub_model
				d["sub_vendor"] = device.sub_vendor

			p["devices"].append(d)

		p["cpu"] = {
			"arch" : self.arch,
			"vendor" : self.cpu.vendor,
			"model" : self.cpu.model,
			"model_string" : self.cpu.model_string,
			"stepping" : self.cpu.stepping,
			"flags" : self.cpu.flags,
			"speed" : self.cpu.speed,
			"family" : self.cpu.family,
			"count" : self.cpu.count				
		}

		if self.cpu.bogomips:
			p["bogomips"] = self.cpu.bogomips

		p["network"] = {
			"green" : self.network.has_green(),
			"blue" : self.network.has_blue(),
			"orange" : self.network.has_orange(),
			"red" : self.network.has_red(),
		 }

		# Only append hypervisor information if we are virtualized.
		if self.virtual:
			p["hypervisor"] = {
				"vendor" : self.hypervisor.vendor,
			}

		return {
			# Profile version
			"profile_version" : PROFILE_VERSION,

			# Identification and authorization codes
			"public_id" : self.public_id,
			"private_id" : self.private_id,

			# Actual profile data
			"profile" : p,
		}
				
		
	@property
	def arch(self):
		return os.uname()[4]

	@property
	def public_id(self):
		"""
			This returns a globally (hopefully) ID to identify the host
			later (by request) in the database.
		"""
		public_id = self.secret_id
		if not public_id:
			return "0" * 40

		return hashlib.sha1(public_id).hexdigest()

	@property
	def private_id(self):
		"""
			The private ID is built out of the _unique_id and used to
			permit a host to do changes on the database.

			No one could ever guess this without access to the host.
		"""
		private_id = ""
		for i in reversed(self.secret_id):
			private_id += i

		if not private_id:
			return "0" * 40

		return hashlib.sha1(private_id).hexdigest()

	@property
	def secret_id(self):
		"""
			Read a "secret" ID from a file if available
			or calculate it from the hardware.
		"""
		if os.path.exists(SECRET_ID_FILE):
			return read_from_file(SECRET_ID_FILE)

		return hashlib.sha1(self._unique_id).hexdigest()

	@property
	def _unique_id(self):
		"""
			This is a helper ID which is generated out of some hardware information
			that is considered to be constant over a PC's lifetime.

			None of the data here is ever sent to the server.
		"""
		ids = []

		# Virtual machines (for example) and some boards have a UUID
		# which is globally unique.
		for file in ("product_uuid", "product_serial", "chassis_serial"):
			id = read_from_file(os.path.join(SYS_CLASS_DMI, file))
			ids.append(id)

		# Sort out all bogous or invalid strings from the list.
		_ids = []
		for id in ids:
			if id is None:
				continue

			for i in INVALID_ID_STRINGS_EXACT_MATCH:
				if id == i:
					id = None
					break

			if id:
				for i in INVALID_ID_STRINGS:
					if i in id:
						id = None
						break

			# Check if the string only contains 0xff
			if id and all((e == "\xff" for e in id)):
				id = None

			if id:
				_ids.append(id)

		ids = _ids

		# Use serial number from root disk (if available) and if
		# no other ID was found, yet.
		if not ids:
			root_disk_serial = self.root_disk_serial
			if root_disk_serial and not root_disk_serial.startswith("QM000"):
				ids.append(root_disk_serial)

		# As last resort, we use the UUID from pakfire.
		if not ids:
			id = read_from_file("/opt/pakfire/db/uuid")
			ids.append(id)

		return "#".join(ids)

	@property
	def language(self):
		"""
			Return the language code of IPFire or "unknown" if we cannot get it.
		"""
		# Return "unknown" if settings file does not exist.
		filename = "/var/ipfire/main/settings"
		if not os.path.exists(filename):
			return "unknown"

		with open(filename, "r") as f:
			for line in f.readlines():
				key, val = line.split("=", 1)
				if key == "LANGUAGE":
					return val.strip()

	@property
	def release(self):
		"""
			Return the system release string.
		"""
		return read_from_file("/etc/system-release") or "unknown"

	@property
	def bios_vendor(self):
		"""
			Return the bios vendor name.
		"""
		return read_from_file("/sys/class/dmi/id/bios_vendor")

	def vendor_model_tuple(self):
		try:
			s = self.__cpuinfo["Hardware"]
		except KeyError:
			return (None, None)

		if s.startswith("ARM-Versatile"):
			return ("ARM", s)

		try:
			v, m = s.split(" ", 1)
		except ValueError:
			if s.startswith("BCM"):
				v = "Broadcom"
				m = s
			else:
				v = None
				m = s

		return v, m

	@staticmethod
	def escape_string(s):
		"""
			Will remove all non-printable characters from the given string
		"""
		if s is None:
			return

		return [x for x in s if x in string.printable]

	@property
	def vendor(self):
		"""
			Return the vendor string of this system (if any).
		"""
		ret = None
		for file in ("sys_vendor", "board_vendor", "chassis_vendor",):
			ret = read_from_file(os.path.join(SYS_CLASS_DMI, file))
			if ret:
				return self.escape_string(ret)

		if os.path.exists("/proc/device-tree"):
			ret = self.__cpuinfo.get("Hardware", None)
		else:
			ret, m = self.vendor_model_tuple()

		return self.escape_string(ret)

	@property
	def model(self):
		"""
			Return the model string of this system (if any).
		"""
		ret = None
		for file in ("product_name", "board_model", "chassis_model",):
			ret = read_from_file(os.path.join(SYS_CLASS_DMI, file))
			if ret:
				return self.escape_string(ret)

		# Read device-tree model if available
		ret = read_from_file("/proc/device-tree/model")
		if ret:
			# replace the NULL byte with which the DT string ends
			ret = ret.replace("\u0000", "")

		# Fall back to read /proc/cpuinfo
		if not ret:
			v, ret = self.vendor_model_tuple()

		return self.escape_string(ret)

	@property
	def memory(self):
		"""
			Return the amount of memory in kilobytes.
		"""
		with open("/proc/meminfo", "r") as f:
			firstline = f.readline().strip()
			return int(firstline.split()[1])

	@property
	def kernel_release(self):
		"""
			Return the kernel release string.
		"""
		return os.uname()[2]

	@property
	def root_disk(self):
		"""
			Return the dev node of the root disk.
		"""
		with open("/proc/mounts", "r") as f:
			for line in f.readlines():
				# Skip empty lines
				if not line:
					continue

				dev, mountpoint, fs, rest = line.split(" ", 3)
				if mountpoint == "/" and not fs == "rootfs":
					# Cut off /dev
					dev = dev[5:]

					# Handle raids and MMC cards like (mmcblk0p3).
					if dev[-2] == "p":
						return dev[:-2]

					# Otherwise cut off all digits at end of string
					while dev[-1] in string.digits:
						dev = dev[:-1]

					return dev

	@property
	def root_size(self):
		"""
			Return the size of the root disk in kilobytes.
		"""
		path = "/sys/block/%s/size" % self.root_disk
		if not os.path.exists(path):
			return

		with open(path, "r") as f:
			return int(f.readline()) * 512 / 1024

	@property
	def root_disk_serial(self):
		"""
			Return the serial number of the root disk (if any).
		"""
		try:
			serial = _fireinfo.get_harddisk_serial("/dev/%s" % self.root_disk)
		except OSError:
			return

		if serial:
			# Strip all spaces
			return serial.strip()

	def scan(self):
		"""
			Scan for all devices (PCI/USB) in the system and append them
			to our list.
		"""
		self.devices = []

		toscan = (
			("/sys/bus/pci/devices", device.PCIDevice),
			("/sys/bus/usb/devices", device.USBDevice)
		)
		for path, cls in toscan:
			if not os.path.exists(path):
				continue

			dirlist = os.listdir(path)
			for dir in dirlist:
				self.devices.append(cls(os.path.join(path, dir)))

	@property
	def virtual(self):
		"""
			Say if the host is running in a virtual environment.
		"""
		return self.hypervisor.virtual

	@property
	def network(self):
		"""
			Reference to the network class.
		"""
		return network.Network()


if __name__ == "__main__":
	s=System()
	print(s.arch)
	print(s.language)
	print(s.release)
	print(s.bios_vendor)
	print(s.memory)
	print(s.kernel)
	print(s.root_disk)
	print(s.root_size)
	print("------------\n", s.devices, "\n------------\n")
	print(json.dumps(s.profile(), sort_keys=True, indent=4))