re_taint_free = re.compile(r"^[-+~/\.\w]+$")
re_parse_maintainer = re.compile(r"^\s*(\S.*\S)\s*\<([^\>]+)\>")
+re_gpg_uid = re.compile('^uid.*<([^>]*)>')
re_srchasver = re.compile(r"^(\S+)\s+\((\S+)\)$")
+re_verwithext = re.compile(r"^(\d+)(?:\.(\d+))(?:\s+\((\S+)\))?$")
changes_parse_error_exc = "Can't parse line in .changes file"
invalid_dsc_format_exc = "Invalid .dsc file"
default_config = "/etc/dak/dak.conf"
default_apt_config = "/etc/dak/apt.conf"
+alias_cache = None
+key_uid_email_cache = {}
+
################################################################################
class Error(Exception):
if section.find('/') != -1:
component = section.split('/')[0]
- if component.lower() == "non-us" and section.find('/') != -1:
- s = component + '/' + section.split('/')[1]
- if Cnf.has_key("Component::%s" % s): # Avoid e.g. non-US/libs
- component = s
-
- if section.lower() == "non-us":
- component = "non-US/main"
-
- # non-US prefix is case insensitive
- if component.lower()[:6] == "non-us":
- component = "non-US"+component[6:]
# Expand default component
if component == "":
component = section
else:
component = "main"
- elif component == "non-US":
- component = "non-US/main"
return (section, component)
# Dropped support for 1.4 and ``buggy dchanges 3.4'' (?!) compared to di.pl
-def build_file_list(changes, is_a_dsc=0):
+def build_file_list(changes, is_a_dsc=0, field="files", hashname="md5sum"):
files = {}
# Make sure we have a Files: field to parse...
- if not changes.has_key("files"):
- raise no_files_exc
+ if not changes.has_key(field):
+ raise no_files_exc
# Make sure we recognise the format of the Files: field
- format = changes.get("format", "")
- if format != "":
- format = float(format)
- if not is_a_dsc and (format < 1.5 or format > 2.0):
- raise nk_format_exc, format
+ format = re_verwithext.search(changes.get("format", "0.0"))
+ if not format:
+ raise nk_format_exc, "%s" % (changes.get("format","0.0"))
+
+ format = format.groups()
+ if format[1] == None:
+ format = int(float(format[0])), 0, format[2]
+ else:
+ format = int(format[0]), int(format[1]), format[2]
+ if format[2] == None:
+ format = format[:2]
+
+ if is_a_dsc:
+ if format != (1,0):
+ raise nk_format_exc, "%s" % (changes.get("format","0.0"))
+ else:
+ if (format < (1,5) or format > (1,8)):
+ raise nk_format_exc, "%s" % (changes.get("format","0.0"))
+ if field != "files" and format < (1,8):
+ raise nk_format_exc, "%s" % (changes.get("format","0.0"))
+
+ includes_section = (not is_a_dsc) and field == "files"
# Parse each entry/line:
- for i in changes["files"].split('\n'):
+ for i in changes[field].split('\n'):
if not i:
break
s = i.split()
section = priority = ""
try:
- if is_a_dsc:
- (md5, size, name) = s
- else:
+ if includes_section:
(md5, size, section, priority, name) = s
+ else:
+ (md5, size, name) = s
except ValueError:
raise changes_parse_error_exc, i
(section, component) = extract_component_from_section(section)
- files[name] = Dict(md5sum=md5, size=size, section=section,
+ files[name] = Dict(size=size, section=section,
priority=priority, component=component)
+ files[name][hashname] = md5
return files
def poolify (source, component):
if component:
component += '/'
- # FIXME: this is nasty
- component = component.lower().replace("non-us/", "non-US/")
if source[:3] == "lib":
return component + source[:4] + '/' + source + '/'
else:
else:
return default_apt_config
+def which_alias_file():
+ hostname = socket.gethostbyaddr(socket.gethostname())[0]
+ aliasfn = '/var/lib/misc/'+hostname+'/forward-alias'
+ if os.path.exists(aliasfn):
+ return aliasfn
+ else:
+ return None
+
################################################################################
# Escape characters which have meaning to SQL's regex comparison operator ('~')
reject("no signature found in %s." % (sig_filename))
bad = 1
if keywords.has_key("KEYEXPIRED") and not keywords.has_key("GOODSIG"):
+ args = keywords["KEYEXPIRED"]
+ if len(args) >= 1:
+ key = args[0]
reject("The key (0x%s) used to sign %s has expired." % (key, sig_filename))
bad = 1
# Finally ensure there's not something we don't recognise
known_keywords = Dict(VALIDSIG="",SIG_ID="",GOODSIG="",BADSIG="",ERRSIG="",
SIGEXPIRED="",KEYREVOKED="",NO_PUBKEY="",BADARMOR="",
- NODATA="")
+ NODATA="",NOTATION_DATA="",NOTATION_NAME="",KEYEXPIRED="")
for keyword in keywords.keys():
if not known_keywords.has_key(keyword):
################################################################################
+def gpg_get_key_addresses(fingerprint):
+ """retreive email addresses from gpg key uids for a given fingerprint"""
+ addresses = key_uid_email_cache.get(fingerprint)
+ if addresses != None:
+ return addresses
+ addresses = set()
+ cmd = "gpg --no-default-keyring %s --fingerprint %s" \
+ % (gpg_keyring_args(), fingerprint)
+ (result, output) = commands.getstatusoutput(cmd)
+ if result == 0:
+ for l in output.split('\n'):
+ m = re_gpg_uid.match(l)
+ if m:
+ addresses.add(m.group(1))
+ key_uid_email_cache[fingerprint] = addresses
+ return addresses
+
+################################################################################
+
# Inspired(tm) by http://www.zopelabs.com/cookbook/1022242603
def wrap(paragraph, max_length, prefix=""):
################################################################################
+# checks if the user part of the email is listed in the alias file
+
+def is_email_alias(email):
+ global alias_cache
+ if alias_cache == None:
+ aliasfn = which_alias_file()
+ alias_cache = set()
+ if aliasfn:
+ for l in open(aliasfn):
+ alias_cache.add(l.split(':')[0])
+ uid = email.split('@')[0]
+ return uid in alias_cache
+
+################################################################################
+
apt_pkg.init()
Cnf = apt_pkg.newConfiguration()