Skip to content

GitLab

Commit 1e4ff9d4 authored by Ondřej Surý's avatar Ondřej Surý Committed by Ondřej Surý
Browse files
Options

Make the python code pylint and flake8 compliant

parent ee534592
Hide whitespace changes
Inline Side-by-side
Showing with 175 additions and 139 deletions
bin/tests/system/digdelv/yamlget.py
View file @ 1e4ff9d4
......@@ -9,25 +9,22 @@
# information regarding copyright ownership.
############################################################################
import sys
try:
import yaml
except:
except (ModuleNotFoundError, ImportError):
print("No python yaml module, skipping")
exit(1)
import subprocess
import pprint
import sys
sys.exit(1)
f = open(sys.argv[1], "r")
for item in yaml.safe_load_all(f):
for key in sys.argv[2:]:
try:
key = int(key)
except: pass
try:
item = item[key]
except:
print('error: index not found')
exit(1)
print (item)
with open(sys.argv[1], "r") as f:
for item in yaml.safe_load_all(f):
for key in sys.argv[2:]:
try:
key = int(key)
except ValueError:
pass
if key not in item:
print('error: index not found')
sys.exit(1)
print(item)
bin/tests/system/dnstap/ydump.py
View file @ 1e4ff9d4
......@@ -9,18 +9,21 @@
# information regarding copyright ownership.
############################################################################
import sys
try:
import yaml
except:
except (ModuleNotFoundError, ImportError):
print("No python yaml module, skipping")
exit(1)
sys.exit(1)
import subprocess
import pprint
import sys
DNSTAP_READ=sys.argv[1]
DATAFILE=sys.argv[2]
DNSTAP_READ = sys.argv[1]
DATAFILE = sys.argv[2]
ARGS = [DNSTAP_READ, '-y', DATAFILE]
f = subprocess.Popen([DNSTAP_READ, '-y', DATAFILE], stdout=subprocess.PIPE)
pprint.pprint([l for l in yaml.load_all(f.stdout)])
with subprocess.Popen(ARGS, stdout=subprocess.PIPE) as f:
for l in yaml.load_all(f.stdout):
pprint.pprint(l)
bin/tests/system/keymgr/testpolicy.py
View file @ 1e4ff9d4
......@@ -10,31 +10,30 @@
############################################################################
import sys
sys.path.insert(0, '../../../python')
from isc import *
from isc import policy
pp = policy.dnssec_policy()
PP = policy.dnssec_policy()
# print the unmodified default and a generated zone policy
print(pp.named_policy['default'])
print(pp.named_policy['global'])
print(pp.policy('example.com'))
print(PP.named_policy['default'])
print(PP.named_policy['global'])
print(PP.policy('example.com'))
if len(sys.argv) > 0:
for policy_file in sys.argv[1:]:
pp.load(policy_file)
PP.load(policy_file)
# now print the modified default and generated zone policies
print(pp.named_policy['default'])
print(pp.policy('example.com'))
print(pp.policy('example.org'))
print(pp.policy('example.net'))
print(PP.named_policy['default'])
print(PP.policy('example.com'))
print(PP.policy('example.org'))
print(PP.policy('example.net'))
# print algorithm policies
print(pp.alg_policy['RSASHA1'])
print(pp.alg_policy['RSASHA256'])
print(pp.alg_policy['ECDSAP256SHA256'])
print(PP.alg_policy['RSASHA1'])
print(PP.alg_policy['RSASHA256'])
print(PP.alg_policy['ECDSAP256SHA256'])
# print another named policy
print(pp.named_policy['extra'])
print(PP.named_policy['extra'])
else:
print("ERROR: Please provide an input file")
bin/tests/system/keymgr/tests.sh
View file @ 1e4ff9d4
......@@ -133,7 +133,7 @@ n=`expr $n + 1`
echo_i "checking policy.conf parser ($n)"
ret=0
${PYTHON} testpolicy.py policy.sample > policy.out
PYTHONPATH="../../../python:$PYTHONPATH" ${PYTHON} testpolicy.py policy.sample > policy.out
$DOS2UNIX policy.out > /dev/null 2>&1
cmp -s policy.good policy.out || ret=1
if [ $ret != 0 ]; then echo_i "failed"; fi
......
contrib/kasp/kasp2policy.py
View file @ 1e4ff9d4
......@@ -21,17 +21,21 @@
from xml.etree import cElementTree as ET
from collections import defaultdict
from isc import dnskey
import re
import ply.yacc as yacc
import ply.lex as lex
import re
from isc import dnskey
############################################################################
# Translate KASP duration values into seconds
############################################################################
class kasptime:
class ktlex:
tokens = ( 'P', 'T', 'Y', 'M', 'D', 'H', 'S', 'NUM' )
class KaspTime:
# pylint: disable=invalid-name
class KTLex:
# pylint: disable=invalid-name
tokens = ('P', 'T', 'Y', 'M', 'D', 'H', 'S', 'NUM')
t_P = r'(?i)P'
t_T = r'(?i)T'
......@@ -41,12 +45,14 @@ class kasptime:
t_H = r'(?i)H'
t_S = r'(?i)S'
def t_NUM(self, t):
@staticmethod
def t_NUM(t):
r'\d+'
t.value = int(t.value)
return t
def t_error(self, t):
@staticmethod
def t_error(t):
print("Illegal character '%s'" % t.value[0])
t.lexer.skip(1)
......@@ -54,7 +60,7 @@ class kasptime:
self.lexer = lex.lex(object=self)
def __init__(self):
self.lexer = self.ktlex()
self.lexer = self.KTLex()
self.tokens = self.lexer.tokens
self.parser = yacc.yacc(debug=False, write_tables=False, module=self)
......@@ -62,35 +68,43 @@ class kasptime:
self.lexer.lexer.lineno = 0
return self.parser.parse(text)
def p_ktime_4(self, p):
@staticmethod
def p_ktime_4(p):
"ktime : P periods T times"
p[0] = p[2] + p[4]
def p_ktime_3(self, p):
@staticmethod
def p_ktime_3(p):
"ktime : P T times"
p[0] = p[3]
def p_ktime_2(self, p):
@staticmethod
def p_ktime_2(p):
"ktime : P periods"
p[0] = p[2]
def p_periods_1(self, p):
@staticmethod
def p_periods_1(p):
"periods : period"
p[0] = p[1]
def p_periods_2(self, p):
@staticmethod
def p_periods_2(p):
"periods : periods period"
p[0] = p[1] + p[2]
def p_times_1(self, p):
@staticmethod
def p_times_1(p):
"times : time"
p[0] = p[1]
def p_times_2(self, p):
@staticmethod
def p_times_2(p):
"times : times time"
p[0] = p[1] + p[2]
def p_period(self, p):
@staticmethod
def p_period(p):
'''period : NUM Y
| NUM M
| NUM D'''
......@@ -101,7 +115,8 @@ class kasptime:
elif p[2].lower() == 'd':
p[0] += int(p[1]) * 86400
def p_time(self, p):
@staticmethod
def p_time(p):
'''time : NUM H
| NUM M
| NUM S'''
......@@ -112,24 +127,28 @@ class kasptime:
elif p[2].lower() == 's':
p[0] = int(p[1])
def p_error(self, p):
@staticmethod
def p_error():
print("Syntax error")
############################################################################
# Load the contents of a KASP XML file as a python dictionary
############################################################################
class kasp():
class Kasp():
# pylint: disable=invalid-name
@staticmethod
def _todict(t):
d = {t.tag: {} if t.attrib else None}
children = list(t)
if children:
dd = defaultdict(list)
for dc in map(kasp._todict, children):
for dc in map(Kasp._todict, children):
for k, v in dc.iteritems():
dd[k].append(v)
d = {t.tag:
{k:v[0] if len(v) == 1 else v for k, v in dd.iteritems()}}
k = {k: v[0] if len(v) == 1 else v for k, v in dd.items()}
d = {t.tag: k}
if t.attrib:
d[t.tag].update(('@' + k, v) for k, v in t.attrib.iteritems())
if t.text:
......@@ -142,7 +161,7 @@ class kasp():
return d
def __init__(self, filename):
self._dict = kasp._todict(ET.parse(filename).getroot())
self._dict = Kasp._todict(ET.parse(filename).getroot())
def __getitem__(self, key):
return self._dict[key]
......@@ -156,52 +175,54 @@ class kasp():
def __repr__(self):
return repr(self._dict)
############################################################################
# Load the contents of a KASP XML file as a python dictionary
############################################################################
if __name__ == "__main__":
from pprint import *
import sys
if len(sys.argv) < 2:
print("Usage: kasp2policy <filename>")
exit(1)
sys.exit(1)
KINFO = Kasp(sys.argv[1])
try:
kinfo = kasp(sys.argv[1])
except:
KINFO = Kasp(sys.argv[1])
except FileNotFoundError:
print("%s: unable to load KASP file '%s'" % (sys.argv[0], sys.argv[1]))
exit(1)
sys.exit(1)
kt = kasptime()
first = True
KT = KaspTime()
FIRST = True
for p in kinfo['KASP']['Policy']:
if not p['@name'] or not p['Keys']: continue
if not first:
for policy in KINFO['KASP']['Policy']:
if not policy['@name'] or not policy['Keys']:
continue
if not FIRST:
print("")
first = False
if p['Description']:
d = p['Description'].strip()
print("# %s" % re.sub(r"\n\s*", "\n# ", d))
print("policy %s {" % p['@name'])
ksk = p['Keys']['KSK']
zsk = p['Keys']['ZSK']
FIRST = False
if policy['Description']:
desc = policy['Description'].strip()
print("# %s" % re.sub(r"\n\s*", "\n# ", desc))
print("policy %s {" % policy['@name'])
ksk = policy['Keys']['KSK']
zsk = policy['Keys']['ZSK']
kalg = ksk['Algorithm']
zalg = zsk['Algorithm']
algnum = kalg['#text'] or zalg['#text']
if algnum:
print("\talgorithm %s;" % dnskey.algstr(int(algnum)))
if p['Keys']['TTL']:
print("\tkeyttl %d;" % kt.parse(p['Keys']['TTL']))
if policy['Keys']['TTL']:
print("\tkeyttl %d;" % KT.parse(policy['Keys']['TTL']))
if kalg['@length']:
print("\tkey-size ksk %d;" % int(kalg['@length']))
if zalg['@length']:
print("\tkey-size zsk %d;" % int(zalg['@length']))
if ksk['Lifetime']:
print("\troll-period ksk %d;" % kt.parse(ksk['Lifetime']))
print("\troll-period ksk %d;" % KT.parse(ksk['Lifetime']))
if zsk['Lifetime']:
print("\troll-period zsk %d;" % kt.parse(zsk['Lifetime']))
print("\troll-period zsk %d;" % KT.parse(zsk['Lifetime']))
if ksk['Standby']:
print("\tstandby ksk %d;" % int(ksk['Standby']))
if zsk['Standby']:
......
contrib/scripts/catzhash.py
View file @ 1e4ff9d4
......@@ -15,16 +15,18 @@
# "domain.example" can be represented in a catalog zone called
# "catalog.example" by adding the following record:
#
# 5960775ba382e7a4e09263fc06e7c00569b6a05c.zones.catalog.example. IN PTR domain.example.
# 5960775ba382e7a4e09263fc06e7c00569b6a05c.zones.catalog.example. \
# IN PTR domain.example.
#
# The label "5960775ba382e7a4e09263fc06e7c00569b6a05c" is the output of
# this script when run with the argument "domain.example".
import sys
import dns.name
import hashlib
import dns.name
if len(sys.argv) < 2:
print("Usage: %s name" % sys.argv[0])
print (hashlib.sha1(dns.name.from_text(sys.argv[1]).to_wire()).hexdigest())
NAME = dns.name.from_text(sys.argv[1]).to_wire()
print(hashlib.sha1(NAME).hexdigest())
util/parse_tsan.py
View file @ 1e4ff9d4
......@@ -10,10 +10,21 @@
# information regarding copyright ownership.
############################################################################
import sys, os, os.path, re
"""Parse the ThreadSanizer reports, unify them and put them into unique dirs."""
import sys
import os
import os.path
import re
from hashlib import sha256
class State:
"""Class that holds state of the TSAN parser."""
# pylint: disable=too-many-instance-attributes
# pylint: disable=too-few-public-methods
inside = False
block = ""
last_line = None
......@@ -25,10 +36,12 @@ class State:
pointers = {}
p_index = 1
def init(self):
def __init__(self):
self.reset()
def reset(self):
"""Reset the object to initial state"""
self.inside = False
self.block = ""
......@@ -41,72 +54,73 @@ class State:
self.t_index = 1
self.p_index = 1
top = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
out = os.path.join(top, "tsan")
TOP = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
if not os.path.isdir(out):
os.mkdir(out)
OUT = os.path.join(TOP, "tsan")
# Regular Expressions
mutex = re.compile(r"M\d+")
thread = re.compile(r"T\d+")
stack = re.compile(r"\s\(\S+\+0x\S+\)")
pointer = re.compile(r"0x[0-9a-f]+")
pid = re.compile(r"\(pid=\d+,?\)")
tid = re.compile(r"tid=\d+,?\s*")
worker = re.compile(r"\s+'(isc-worker|isc-net-)\d+'")
path = re.compile(top + "/")
if not os.path.isdir(OUT):
os.mkdir(OUT)
s = State()
# Regular Expressions
MUTEX = re.compile(r"M\d+")
THREAD = re.compile(r"T\d+")
STACK = re.compile(r"\s\(\S+\+0x\S+\)")
POINTER = re.compile(r"0x[0-9a-f]+")
PID = re.compile(r"\(pid=\d+,?\)")
TID = re.compile(r"tid=\d+,?\s*")
WORKER = re.compile(r"\s+'(isc-worker|isc-net-)\d+'")
PATH = re.compile(TOP + "/")
S = State()
with open(sys.argv[1], "r", encoding='utf-8') as f:
lines = f.readlines()
for line in lines:
for line in f.readlines():
if line == "==================\n":
if not s.inside:
s.inside = True
else:
dname = os.path.join(out, sha256(s.last_line.encode('utf-8')).hexdigest())
if not os.path.isdir(dname):
os.mkdir(dname)
fname = os.path.join(dname, sha256(s.block.encode('utf-8')).hexdigest() + ".tsan")
if not os.path.isfile(fname):
with open(fname, "w", encoding='utf-8') as w:
w.write(s.block)
s.reset()
if not S.inside:
S.inside = True
else:
dname = sha256(S.last_line.encode('utf-8')).hexdigest()
dname = os.path.join(OUT, dname)
if not os.path.isdir(dname):
os.mkdir(dname)
fname = sha256(S.block.encode('utf-8')).hexdigest() + ".tsan"
fname = os.path.join(dname, fname)
if not os.path.isfile(fname):
with open(fname, "w", encoding='utf-8') as w:
w.write(S.block)
S.reset()
else:
for m in mutex.finditer(line):
for m in MUTEX.finditer(line):
k = m.group()
if k not in s.mutexes:
s.mutexes[k] = s.m_index
s.m_index += 1
for m in thread.finditer(line):
if k not in S.mutexes:
S.mutexes[k] = S.m_index
S.m_index += 1
for m in THREAD.finditer(line):
k = m.group()
if k not in s.threads:
s.threads[k] = s.t_index
s.t_index += 1
for m in pointer.finditer(line):
if k not in S.threads:
S.threads[k] = S.t_index
S.t_index += 1
for m in POINTER.finditer(line):
k = m.group()
if k not in s.pointers:
s.pointers[k] = s.p_index
s.p_index += 1
for k, v in s.mutexes.items():
if k not in S.pointers:
S.pointers[k] = S.p_index
S.p_index += 1
for k, v in S.mutexes.items():
r = re.compile(k)
line = r.sub("M%s" % v, line)
for k, v in s.threads.items():
for k, v in S.threads.items():
r = re.compile(k)
line = r.sub("T%s" % v, line)
for k, v in s.pointers.items():
for k, v in S.pointers.items():
r = re.compile(k)
line = r.sub("0x%s" % str(v).zfill(12), line)
line = stack.sub("", line)
line = pid.sub("", line)
line = tid.sub("", line)
line = worker.sub("", line)
line = path.sub("", line)
line = STACK.sub("", line)
line = PID.sub("", line)
line = TID.sub("", line)
line = WORKER.sub("", line)
line = PATH.sub("", line)
s.block += line
s.last_line = line
S.block += line
S.last_line = line
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment