rewritten glsa class for the new DTD

glsa.py (text/plain), 11.18 KB, created by Marius Mauch (RETIRED) on 2003-12-21 20:00:34 UTC

#!/usr/bin/python2.2
# $Header: /home/cvsroot/gentoo-projects/gentoo-security/GLSA/user-tools/glsa.py,v 1.10 2003/12/21 04:18:06 genone Exp $

# This program is licensed under the GPL, version 2

# WARNING: this code is only tested by a few people and should NOT be used
# on production systems at this stage. There are possible security holes and probably
# bugs in this code. If you test it please report ANY success or failure to
# me ([e-mail]).

# This is the main class for GLSA support in portage. It's currently designed
# to interface with portage through emerge and portageq, this will be changed
# to internal portage versions over time. It's the backend code for the glsa-check.py
# tool, which will go into portage.py / emerge at some point.

# The following planned features are currently on hold:
# - getting GLSAs from http/ftp servers (not really useful without the fixed ebuilds)
# - GPG signing/verification (until key policy is clear)

__author__ = "Marius Mauch <[e-mail]>"

import os, sys, urllib, time, string, portage, codecs
import xml.dom.minidom

sys.path += ["/usr/lib/portage/pym"]	# to find portage.py

opMapping = {"le": "<=", "lt": "<", "eq": "=", "gt": ">", "ge": ">="}
NEWLINE_ESCAPE = "!;\\n"

def center(text, width):
	margin = (width-len(text))/2
	rValue = " "*margin
	rValue += text
	if 2*margin + len(text) == width:
		rValue += " "*margin
	elif 2*margin + len(text) + 1 == width:
		rValue += " "*(margin+1)
	return rValue

def wrap(text, width, caption=""):
	"""
	A word-wrap function that preserves existing line breaks
	and most spaces in the text. Expects that existing line
	breaks are posix newlines (\n).
	"""
	rValue = ""
	line = caption
	words = text.split()
	indentLevel = len(caption)+1
	for w in words:
		if len(line)+len(w)+1 > width:
			rValue += line+"\n"
			line = " "*indentLevel+w
		elif w.find(NEWLINE_ESCAPE) >= 0:
			if len(line.strip()) > 0:
				rValue += line+" "+w.replace(NEWLINE_ESCAPE, "\n")
			else:
				rValue += line+w.replace(NEWLINE_ESCAPE, "\n")
			line = " "*indentLevel
		else:
			if len(line.strip()) > 0:
				line += " "+w
			else:
				line += w
	if len(line) > 0:
		rValue += line.replace(NEWLINE_ESCAPE, "\n")
	return rValue

# take normal portage config and add GLSA stuff if missing
def checkconfig(myconfig):
	mysettings = dict([
		("GLSA_DIR", portage.settings["PORTDIR"]+"/glsa/"),
		("GLSA_PREFIX", "glsa-"),
		("GLSA_SUFFIX", ".xml"),
		("CHECKFILE", "/var/cache/edb/glsa"),
		("GLSA_SERVER", "www.gentoo.org/security/en/glsa/"),	# not completely implemented yet
		("CHECKMODE", "local"),									# not completely implemented yet
		("PRINTWIDTH", "76")
	])
	for k in mysettings.keys():
		if not myconfig.has_key(k):
			myconfig[k] = mysettings[k]
	return myconfig

glsaconfig = checkconfig(portage.config(clone=portage.settings))

# get a list of all available GLSA in the given repository
def get_glsa_list(repository):

	# TODO: remote fetch code for listing

	rValue = []

	if not os.access(repository, os.R_OK):
		return []
	dirlist = os.listdir(repository)
	prefix = glsaconfig["GLSA_PREFIX"]
	suffix = glsaconfig["GLSA_SUFFIX"]
	
	for f in dirlist:
		if f[:len(prefix)] == prefix:
			rValue.append(f[len(prefix):-1*len(suffix)])
	return rValue

def getListElements(listnode):
	rValue = []
	for li in listnode.childNodes:
		rValue.append(getText(li, format="strip"))
	return rValue

def getText(node, format):
	rValue = ""
	if format in ["strip", "keep"]:
		if node.nodeName == "uri":
			rValue += node.childNodes[0].data+": "+node.getAttribute("link")
		else:
			for subnode in node.childNodes:
				try:
					rValue += subnode.data
				except AttributeError:
					rValue += getText(subnode, format)
	else:
		for subnode in node.childNodes:
			if subnode.nodeName == "p":
				for p_subnode in subnode.childNodes:
					if p_subnode.nodeName == "#text":
						rValue += p_subnode.data
					elif p_subnode.nodeName in ["uri", "mail"]:
						rValue += p_subnode.childNodes[0].data
						rValue += " ( "+p_subnode.getAttribute("link")+" )"
				rValue += NEWLINE_ESCAPE
			elif subnode.nodeName == "ul":
				for li in getListElements(subnode):
					rValue += "- "+li+NEWLINE_ESCAPE+" "
			elif subnode.nodeName == "ol":
				i = 0
				for li in getListElements(subnode):
					i = i+1
					rValue += str(i)+". "+li+NEWLINE_ESCAPE+" "
			elif subnode.nodeName == "code":
				rValue += getText(subnode, format="keep").replace("\n", NEWLINE_ESCAPE)
			elif subnode.nodeName == "#text":
				rValue += subnode.data
			else:
				raise Exception("Unallowed Tag found: ", subnode.nodeName)
	if format == "strip":
		rValue = rValue.strip(" \n\t")
		rValue = rValue.replace("\n", " ")
		rValue = rValue.replace("\t", " ")
		while rValue.find("  ") >= 0:
			rValue = rValue.replace("  ", " ")
	return rValue

def getMultiTagsText(rootnode, tagname, format):
	rValue = []
	for e in rootnode.getElementsByTagName(tagname):
		rValue.append(getText(e, format))
	return rValue

def makeAtom(pkgname, versionNode):
	return opMapping[versionNode.getAttribute("range")] \
			+pkgname \
			+"-"+getText(versionNode, format="strip")

def makeVersion(versionNode):
	return opMapping[versionNode.getAttribute("range")] \
			+getText(versionNode, format="strip")

def getMinUpgrade(vulnerableList, unaffectedList):
	rValue = None
	for v in vulnerableList:
		installed = portage.db["/"]["vartree"].dbapi.match(v)
		if not installed:
			continue
		for u in unaffectedList:
			for c in portage.db["/"]["porttree"].dbapi.match(u):
				c_pv = portage.catpkgsplit(c)
				i_pv = portage.catpkgsplit(portage.best(installed))
				if portage.pkgcmp(c_pv[1:], i_pv[1:]) > 0 and (rValue == None or portage.pkgcmp(c_pv[1:], rValue) < 0):
					rValue = c_pv[0]+"/"+c_pv[1]+"-"+c_pv[2]
					if c_pv[3] != "r0":
						rValue += "-"+c_pv[3]
	return rValue
				
# GLSA xml data wrapper class
class glsa:
	# set the id and read the xml file
	def __init__(self, myid):
		self.nr = myid
		self.read()

	# read the xml file for this glsa using GLSA_DIR if CHECKMODE=local or GLSA_SERVER otherwise
	def read(self):
		if glsaconfig["CHECKMODE"] == "local":
			repository = "file://" + glsaconfig["GLSA_DIR"]
		else:
			repository = glsaconfig["GLSA_SERVER"]
		myurl = repository + glsaconfig["GLSA_PREFIX"] + str(self.nr) + glsaconfig["GLSA_SUFFIX"]
		self.parse(urllib.urlopen(myurl))

	def parse(self, myfile):
		myroot = xml.dom.minidom.parse(myfile).getElementsByTagName("glsa")[0]
		if myroot.getAttribute("id") != self.nr:
			raise Exception("filename and internal id don't match:" + myroot.getAttribute("id") + " != " + self.nr)

		# the simple (single, required, top-level, #PCDATA) tags first
		self.title = getText(myroot.getElementsByTagName("title")[0], format="strip")
		self.synopsis = getText(myroot.getElementsByTagName("synopsis")[0], format="strip")
		self.announced = getText(myroot.getElementsByTagName("announced")[0], format="strip")
		self.revised = getText(myroot.getElementsByTagName("revised")[0], format="strip")
		
		# now the optional and 0-n topelevel, #PCDATA tags and references
		try:
			self.access = getText(myroot.getElementsByTagName("access")[0], format="strip")
		except IndexError:
			self.access = None
		self.bugs = getMultiTagsText(myroot, "bug", format="strip")
		self.references = getMultiTagsText(myroot.getElementsByTagName("references")[0], "uri", format="keep")
		
		# and now the formatted text elements
		self.description = getText(myroot.getElementsByTagName("description")[0], format="xml")
		self.workaround = getText(myroot.getElementsByTagName("workaround")[0], format="xml")
		self.resolution = getText(myroot.getElementsByTagName("resolution")[0], format="xml")
		self.impact_text = getText(myroot.getElementsByTagName("impact")[0], format="xml")
		self.impact_type = myroot.getElementsByTagName("impact")[0].getAttribute("type")
		try:
			self.background = getText(myroot.getElementsByTagName("background")[0], format="xml")
		except IndexError:
			self.background = None					

		# finally the interesting tags (product, affected, package)
		self.glsatype = myroot.getElementsByTagName("product")[0].getAttribute("type")
		self.product = getText(myroot.getElementsByTagName("product")[0], format="strip")
		self.affected = myroot.getElementsByTagName("affected")[0]
		self.packages = {}
		for p in self.affected.getElementsByTagName("package"):
			name = p.getAttribute("name")
			self.packages[name] = {}
			self.packages[name]["arch"] = p.getAttribute("arch")
			self.packages[name]["auto"] = (p.getAttribute("auto") == "yes")
			self.packages[name]["vul_vers"] = [makeVersion(v) for v in p.getElementsByTagName("vulnerable")]
			self.packages[name]["unaff_vers"] = [makeVersion(v) for v in p.getElementsByTagName("unaffected")]
			self.packages[name]["vul_atoms"] = [makeAtom(name, v) for v in p.getElementsByTagName("vulnerable")]
			self.packages[name]["unaff_atoms"] = [makeAtom(name, v) for v in p.getElementsByTagName("unaffected")]
		# TODO: services aren't really used yet
		self.services = self.affected.getElementsByTagName("service")

	def dump(self, outfile="/dev/stdout"):
		myfile = codecs.open(outfile, "w", sys.getfilesystemencoding())
		width = int(glsaconfig["PRINTWIDTH"])
		myfile.write(center("GLSA %s: %s" % (self.nr, self.title), width)+"\n")
		myfile.write((width*"=")+"\n")
		myfile.write(wrap(self.synopsis, width, caption="Synopsis:         ")+"\n")
		myfile.write("Announced on:      %s\n" % self.announced)
		myfile.write("Last revised on:   %s\n\n" % self.revised)
		if self.glsatype == "ebuild":
			for pkg in self.packages.keys():
				vul_vers = string.join(self.packages[pkg]["vul_vers"])
				unaff_vers = string.join(self.packages[pkg]["unaff_vers"])
				myfile.write("Affected package:  %s\n" % pkg)
				myfile.write("Vulnerable:        %s\n" % vul_vers)
				myfile.write("Unaffected:        %s\n" % unaff_vers)
		elif self.glsatype == "infrastructure":
			pass
		if len(self.bugs) > 0:
			myfile.write("\nRelated bugs:      ")
			for i in range(0, len(self.bugs)):
				myfile.write(self.bugs[i])
				if i < len(self.bugs)-1:
					myfile.write(", ")
				else:
					myfile.write("\n")				
		if self.background:
			myfile.write("\n"+wrap(self.background, width, caption="Background:       "))
		myfile.write("\n"+wrap(self.description, width, caption="Description:      "))
		myfile.write("\n"+wrap(self.impact_text, width, caption="Impact:           "))
		myfile.write("\n"+wrap(self.workaround, width, caption="Workaround:       "))
		myfile.write("\n"+wrap(self.resolution, width, caption="Resolution:       "))
		myfile.write("\nReferences:        ")
		for r in self.references:
			myfile.write(r+"\n"+19*" ")
		myfile.write("\n")
		myfile.close()
	
	def isVulnerable(self):
		vList = []
		rValue = False
		for pkg in self.packages.keys():
			vList += self.packages[pkg]["vul_atoms"]
		for v in vList:
			rValue = rValue or len(portage.db["/"]["vartree"].dbapi.match(v)) > 0
		return rValue
	
	def isApplied(self):
		aList = portage.grabfile(glsaconfig["CHECKFILE"])
		return self.nr in aList

	def inject(self):
		if not self.isApplied():
			checkfile = open(glsaconfig["CHECKFILE"], "r+")
			checkfile.write(self.nr+"\n")
			checkfile.close()
	
	def getMergeList(self):
		rValue = []
		for pkg in self.packages.keys():
			update = getMinUpgrade(self.packages[pkg]["vul_atoms"],
									self.packages[pkg]["unaff_atoms"])
			print pkg, update
			if update:
				rValue.append(update)
		return rValue