<s> import re <EOL> import sys <EOL> import time <EOL> import uuid <EOL> import supybot . gpg as gpg <EOL> import supybot . conf as conf <EOL> import supybot . utils as utils <EOL> import supybot . ircdb as ircdb <EOL> from supybot . commands import * <EOL> import supybot . utils . minisix as minisix <EOL> import supybot . plugins as plugins <EOL> import supybot . ircutils as ircutils <EOL> import supybot . callbacks as callbacks <EOL> try : <EOL> from supybot . i18n import PluginInternationalization <EOL> _ = PluginInternationalization ( '<STR_LIT>' ) <EOL> except ImportError : <EOL> _ = lambda x : x <EOL> class GPG ( callbacks . Plugin ) : <EOL> """<STR_LIT>""" <EOL> class key ( callbacks . Commands ) : <EOL> def add ( self , irc , msg , args , user , keyid , keyserver ) : <EOL> """<STR_LIT>""" <EOL> if keyid in user . gpgkeys : <EOL> irc . error ( _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> return <EOL> result = gpg . keyring . recv_keys ( keyserver , keyid ) <EOL> reply = format ( _ ( '<STR_LIT>' ) , <EOL> ( result . imported , _ ( '<STR_LIT:key>' ) ) , <EOL> result . unchanged , <EOL> result . not_imported , <EOL> [ x [ '<STR_LIT>' ] for x in result . results ] ) <EOL> if result . imported == <NUM_LIT:1> : <EOL> user . gpgkeys . append ( keyid ) <EOL> irc . reply ( reply ) <EOL> else : <EOL> irc . error ( reply ) <EOL> add = wrap ( add , [ '<STR_LIT:user>' , <EOL> ( '<STR_LIT>' , <EOL> _ ( '<STR_LIT>' ) ) , <EOL> ( '<STR_LIT>' , <EOL> _ ( '<STR_LIT>' ) ) ] ) <EOL> def remove ( self , irc , msg , args , user , fingerprint ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> keyids = [ x [ '<STR_LIT>' ] for x in gpg . keyring . list_keys ( ) <EOL> if x [ '<STR_LIT>' ] == fingerprint ] <EOL> if len ( keyids ) == <NUM_LIT:0> : <EOL> raise ValueError <EOL> for keyid in keyids : <EOL> try : <EOL> user . gpgkeys . remove ( keyid ) <EOL> except ValueError : <EOL> user . gpgkeys . remove ( '<STR_LIT>' + keyid ) <EOL> gpg . keyring . delete_keys ( fingerprint ) <EOL> irc . replySuccess ( ) <EOL> except ValueError : <EOL> irc . error ( _ ( '<STR_LIT>' ) ) <EOL> remove = wrap ( remove , [ '<STR_LIT:user>' , '<STR_LIT>' ] ) <EOL> def list ( self , irc , msg , args , user ) : <EOL> """<STR_LIT>""" <EOL> keyids = user . gpgkeys <EOL> if len ( keyids ) == <NUM_LIT:0> : <EOL> irc . reply ( _ ( '<STR_LIT>' ) ) <EOL> else : <EOL> irc . reply ( format ( '<STR_LIT>' , keyids ) ) <EOL> list = wrap ( list , [ '<STR_LIT:user>' ] ) <EOL> class signing ( callbacks . Commands ) : <EOL> def __init__ ( self , * args ) : <EOL> super ( GPG . signing , self ) . __init__ ( * args ) <EOL> self . _tokens = { } <EOL> def _expire_tokens ( self ) : <EOL> now = time . time ( ) <EOL> self . _tokens = dict ( filter ( lambda x_y : x_y [ <NUM_LIT:1> ] [ <NUM_LIT:1> ] > now , <EOL> self . _tokens . items ( ) ) ) <EOL> def gettoken ( self , irc , msg , args ) : <EOL> """<STR_LIT>""" <EOL> self . _expire_tokens ( ) <EOL> token = '<STR_LIT>' % str ( uuid . uuid4 ( ) ) <EOL> lifetime = conf . supybot . plugins . GPG . auth . sign . TokenTimeout ( ) <EOL> self . _tokens . update ( { token : ( msg . prefix , time . time ( ) + lifetime ) } ) <EOL> irc . reply ( _ ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) % token ) <EOL> gettoken = wrap ( gettoken , [ ] ) <EOL> _auth_re = re . compile ( r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' , <EOL> re . S ) <EOL> def auth ( self , irc , msg , args , url ) : <EOL> """<STR_LIT>""" <EOL> self . _expire_tokens ( ) <EOL> content = utils . web . getUrl ( url ) <EOL> if minisix . PY3 and isinstance ( content , bytes ) : <EOL> content = content . decode ( ) <EOL> match = self . _auth_re . search ( content ) <EOL> if not match : <EOL> irc . error ( _ ( '<STR_LIT>' ) , Raise = True ) <EOL> data = match . group ( <NUM_LIT:0> ) <EOL> token = match . group ( <NUM_LIT:1> ) <EOL> if token not in self . _tokens : <EOL> irc . error ( _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , Raise = True ) <EOL> if self . _tokens [ token ] [ <NUM_LIT:0> ] != msg . prefix : <EOL> irc . error ( _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , Raise = True ) <EOL> verified = gpg . keyring . verify ( data ) <EOL> if verified and verified . valid : <EOL> keyid = verified . pubkey_fingerprint [ - <NUM_LIT:16> : ] <EOL> prefix , expiry = self . _tokens . pop ( token ) <EOL> found = False <EOL> for ( id , user ) in ircdb . users . items ( ) : <EOL> if keyid in [ x [ - len ( keyid ) : ] for x in user . gpgkeys ] : <EOL> try : <EOL> user . addAuth ( msg . prefix ) <EOL> except ValueError : <EOL> irc . error ( _ ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , Raise = True ) <EOL> ircdb . users . setUser ( user , flush = False ) <EOL> irc . reply ( _ ( '<STR_LIT>' ) % <EOL> user . name ) <EOL> return <EOL> irc . error ( _ ( '<STR_LIT>' ) , Raise = True ) <EOL> else : <EOL> irc . error ( _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> auth = wrap ( auth , [ '<STR_LIT:url>' ] ) <EOL> Class = GPG </s>
