# -g supplementary group membership
# -p File of Phrases that must be in the plaintext.
# -m Disallow PGP/MIME
+# -v Verbose mode
# Typical Debian invokation may look like:
-# ./gpgwrapper -k /usr/share/keyrings/debian-keyring.gpg:/usr/share/keyrings/debian-keyring.pgp \
+# sigcheck -k /usr/share/keyrings/debian-keyring.gpg:/usr/share/keyrings/debian-keyring.pgp \
# -d ou=users,dc=debian,dc=org -l db.debian.org \
# -m debian.org -a admin@db.debian.org \
# -e /etc/userdir-ldap/templtes/error-reply -- test.sh
import sys, traceback, time, os;
-import string, pwd, getopt;
+import pwd, getopt;
+import email, email.parser
from userdir_gpg import *;
EX_TEMPFAIL = 75;
GroupMember = None;
Phrases = None;
AllowMIME = 1;
+Verbose = 0;
+
+def verbmsg(msg):
+ if Verbose:
+ sys.stderr.write(msg + "\n")
# Match the key fingerprint against an LDAP directory
def CheckLDAP(FingerPrint):
import ldap;
+ import userdir_ldap;
# Connect to the ldap server
global ErrTyp, ErrMsg;
ErrType = EX_TEMPFAIL;
- ErrMsg = "An error occured while performing the LDAP lookup:";
+ ErrMsg = "An error occurred while performing the LDAP lookup:";
global l;
- l = ldap.open(LDAPServer);
+ l = userdir_ldap.connectLDAP(LDAPServer);
l.simple_bind_s("","");
# Search for the matching key fingerprint
+ verbmsg("Processing fingerprint %s" % FingerPrint)
Attrs = l.search_s(LDAPDn,ldap.SCOPE_ONELEVEL,"keyfingerprint=" + FingerPrint);
if len(Attrs) == 0:
raise Error, "Key not found"
if len(Attrs) != 1:
raise Error, "Oddly your key fingerprint is assigned to more than one account.."
+ gidnumber_found = 0;
+ for key in Attrs[0][1].keys():
+ if (key == "gidNumber"):
+ gidnumber_found = 1
+
+ if (gidnumber_found != 1):
+ raise Error, "No gidnumber in attributes for fingerprint %s" % FingerPrint
+
+ # Look for the group with the gid of the user
+ GAttr = l.search_s(LDAPDn,ldap.SCOPE_ONELEVEL,"(&(objectClass=debianGroup)(gidnumber=%s))" % Attrs[0][1]["gidNumber"][0], ["gid"])
+ if len(GAttr) == 0:
+ raise Error, "Database inconsistency found: main group for account not found in database"
+
# See if the group membership is OK
+ # Only if a group was given on the commandline
if GroupMember != None:
Hit = 0;
- for x in Attrs[0][1].get("supplementarygid",[]):
- if x == GroupMember:
- Hit = 1;
+ # Check primary group first
+ if GAttr[0][1]["gid"][0] == GroupMember:
+ Hit = 1
+ else:
+ # Check supplementary groups
+ for x in Attrs[0][1].get("supplementaryGid",[]):
+ if x == GroupMember:
+ Hit = 1;
if Hit != 1:
raise Error, "You don't have %s group permissions."%(GroupMember);
# Start of main program
# Process options
-(options, arguments) = getopt.getopt(sys.argv[1:], "r:k:d:l:g:mp:");
+(options, arguments) = getopt.getopt(sys.argv[1:], "r:k:d:l:g:mp:v");
for (switch, val) in options:
if (switch == '-r'):
ReplayCacheFile = val;
elif (switch == '-k'):
- SetKeyrings(string.split(val,":"));
+ SetKeyrings(val.split(":"));
elif (switch == '-d'):
LDAPDn = val;
elif (switch == '-l'):
GroupMember = val;
elif (switch == '-m'):
AllowMIME = 0;
+ elif (switch == '-v'):
+ Verbose = 1;
elif (switch == '-p'):
Phrases = val;
if ReplayCacheFile != None:
ErrMsg = "Failed to initialize the replay cache:";
RC = ReplayCache(ReplayCacheFile);
- RC.Clean();
-
+
# Get the email
ErrType = EX_PERMFAIL;
ErrMsg = "Failed to understand the email or find a signature:";
- Email = mimetools.Message(sys.stdin,0);
- MsgID = Email.getheader("Message-ID");
- Msg = GetClearSig(Email,1);
+ mail = email.parser.Parser().parse(sys.stdin);
+ MsgID = mail["Message-ID"]
+
+ print "Inspecting message %s"%MsgID;
+ verbmsg("Processing message %s" % MsgID)
+ Msg = GetClearSig(mail,1);
if AllowMIME == 0 and Msg[1] != 0:
raise Error, "PGP/MIME disallowed";
ErrMsg = "Message is not PGP signed:"
- if string.find(Msg[0],"-----BEGIN PGP SIGNED MESSAGE-----") == -1:
+ if Msg[0].find("-----BEGIN PGP SIGNED MESSAGE-----") == -1:
raise Error, "No PGP signature";
# Check the signature
ErrMsg = "Unable to check the signature or the signature was invalid:";
- Res = GPGCheckSig(Msg[0]);
-
- if Res[0] != None:
- raise Error, Res[0];
-
- if Res[3] == None:
- raise Error, "Null signature text";
+ pgp = GPGCheckSig2(Msg[0])
+
+ if not pgp.ok:
+ raise UDFormatError, pgp.why
+ if pgp.text is None:
+ raise UDFormatError, "Null signature text"
# Check the signature against the replay cache
if ReplayCacheFile != None:
- ErrMsg = "The replay cache rejected your message. Check your clock!";
- Rply = RC.Check(Res[1]);
- if Rply != None:
- raise Error, Rply;
- RC.Add(Res[1]);
+ RC.process(pgp.sig_info)
# Do LDAP stuff
if LDAPDn != None:
- CheckLDAP(Res[2][1]);
-
+ CheckLDAP(pgp.key_fpr)
+
ErrMsg = "Verifying message:";
if Phrases != None:
F = open(Phrases,"r");
while 1:
Line = F.readline();
if Line == "": break;
- if string.find(Res[3],string.strip(Line)) == -1:
- raise Error,"Phrase '%s' was not found"%(string.strip(Line));
+ if pgp.text.find(Line.strip()) == -1:
+ raise Error,"Phrase '%s' was not found" % (Line.strip())
except:
ErrMsg = "[%s] \"%s\" \"%s %s\"\n"%(Now,MsgID,ErrMsg,sys.exc_value);
sys.exit(EX_PERMFAIL);
# For Main
+print "Message %s passed"%MsgID;
sys.exit(0);
projects / mirror / userdir-ldap.git / blobdiff