<s> import os <EOL> import ast <EOL> import sys <EOL> import math <EOL> import time <EOL> import string <EOL> import hashlib <EOL> import tempfile <EOL> import subprocess <EOL> from operator import itemgetter <EOL> from contextlib import contextmanager <EOL> from getpass import getpass <EOL> import random ; random = random . SystemRandom ( ) <EOL> import sdb . subprocess_compat as subprocess <EOL> from sdb . util import force_bytes <EOL> from sdb . clipboard import set_clipboard_once , ClipboardException <EOL> from sdb . diceware import WORDS <EOL> from sdb import gpg_agent <EOL> def encode ( records ) : <EOL> res = [ ] <EOL> for record in records : <EOL> res . append ( repr ( record ) ) <EOL> return ( '<STR_LIT:\n>' . join ( res ) + '<STR_LIT:\n>' ) . encode ( '<STR_LIT:utf-8>' ) <EOL> def decode ( str ) : <EOL> records = [ ] <EOL> for line in str . decode ( '<STR_LIT:utf-8>' ) . split ( '<STR_LIT:\n>' ) : <EOL> if line : <EOL> records . append ( ast . literal_eval ( line ) ) <EOL> return records <EOL> CASE_ALPHABET = string . ascii_letters <EOL> ALPHANUMERIC = CASE_ALPHABET + string . digits <EOL> EVERYTHING = ALPHANUMERIC + string . punctuation <EOL> def gen_password ( choices = ALPHANUMERIC , length = <NUM_LIT:10> ) : <EOL> return '<STR_LIT>' . join ( random . choice ( choices ) for i in range ( length ) ) <EOL> def requirements_satisfied ( requirements , str ) : <EOL> return all ( [ i in str for i in requirements ] ) <EOL> def gen_password_require ( requirements , choices = ALPHANUMERIC , length = <NUM_LIT:10> ) : <EOL> """<STR_LIT>""" <EOL> if len ( requirements ) > length or not requirements_satisfied ( requirements , choices ) : <EOL> raise Exception ( <EOL> "<STR_LIT>" % ( <EOL> requirements , choices ) ) <EOL> while True : <EOL> pw = gen_password ( choices , length ) <EOL> if requirements_satisfied ( requirements , pw ) : <EOL> return pw <EOL> def gen_password_entropy ( entropy , choices = ALPHANUMERIC ) : <EOL> """<STR_LIT>""" <EOL> required_length = int ( math . ceil ( entropy / math . log ( len ( choices ) , <NUM_LIT:2> ) ) ) <EOL> return gen_password ( choices = choices , length = required_length ) <EOL> def match ( needle , haystack ) : <EOL> score = <NUM_LIT:1> <EOL> j = <NUM_LIT:0> <EOL> last_match = <NUM_LIT:0> <EOL> for c in needle : <EOL> while j < len ( haystack ) and haystack [ j ] != c : <EOL> j += <NUM_LIT:1> <EOL> if j >= len ( haystack ) : <EOL> return <NUM_LIT:0> <EOL> score += <NUM_LIT:1> / ( last_match + <NUM_LIT:1.> ) <EOL> last_match = j <EOL> j += <NUM_LIT:1> <EOL> return score <EOL> def record_score ( term , records ) : <EOL> return match ( term , records [ <NUM_LIT:0> ] + records [ <NUM_LIT:1> ] + records [ <NUM_LIT:3> ] ) <EOL> def search ( term , records ) : <EOL> records = [ ( record_score ( term , i ) , i ) for i in records ] <EOL> records = list ( filter ( itemgetter ( <NUM_LIT:0> ) , records ) ) <EOL> records . sort ( key = itemgetter ( <NUM_LIT:0> ) , reverse = True ) <EOL> return [ i [ <NUM_LIT:1> ] for i in records ] <EOL> def is_unique_list ( lst ) : <EOL> return len ( lst ) == len ( set ( lst ) ) <EOL> def disambiguate ( records ) : <EOL> choices = [ itemgetter ( <NUM_LIT:0> ) , <EOL> itemgetter ( <NUM_LIT:0> , <NUM_LIT:1> ) , <EOL> itemgetter ( <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:3> ) ] <EOL> for choice in choices : <EOL> result = list ( map ( choice , records ) ) <EOL> if is_unique_list ( result ) : <EOL> return result <EOL> return records <EOL> class GPGException ( Exception ) : <EOL> pass <EOL> class IncorrectPasswordException ( GPGException ) : <EOL> pass <EOL> class InvalidEncryptedFileException ( GPGException ) : <EOL> pass <EOL> class FileCorruptionException ( GPGException ) : <EOL> pass <EOL> def gpg_exception_factory ( returncode , message ) : <EOL> if returncode == <NUM_LIT:2> : <EOL> if b'<STR_LIT>' in message : <EOL> return IncorrectPasswordException ( message ) <EOL> if b'<STR_LIT>' in message : <EOL> return FileCorruptionException ( message ) <EOL> if b'<STR_LIT>' in message : <EOL> return FileCorruptionException ( message ) <EOL> if b'<STR_LIT>' in message : <EOL> return FileCorruptionException ( message ) <EOL> if b'<STR_LIT>' : <EOL> return InvalidEncryptedFileException ( message ) <EOL> return Exception ( "<STR_LIT>" , returncode , message ) <EOL> def dencrypt ( command , pw , data ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT:\n>' in pw : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> proc = subprocess . Popen ( <EOL> command , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE <EOL> ) <EOL> proc . stdin . write ( force_bytes ( pw ) ) <EOL> proc . stdin . write ( b'<STR_LIT:\n>' ) <EOL> proc . stdin . write ( data ) <EOL> output , erroroutput = proc . communicate ( ) <EOL> if proc . returncode != <NUM_LIT:0> : <EOL> raise gpg_exception_factory ( proc . returncode , erroroutput ) <EOL> return output <EOL> def encrypt ( pw , data ) : <EOL> return dencrypt ( <EOL> [ '<STR_LIT>' , '<STR_LIT:-c>' , <EOL> '<STR_LIT>' , '<STR_LIT:0>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] , <EOL> pw , <EOL> data , <EOL> ) <EOL> def decrypt ( pw , data ) : <EOL> return dencrypt ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:0>' , '<STR_LIT>' ] , <EOL> pw , <EOL> data <EOL> ) <EOL> def get_tmp_file ( filename ) : <EOL> file_parts = os . path . split ( filename ) <EOL> return os . path . join ( * file_parts [ : - <NUM_LIT:1> ] + ( '<STR_LIT:.>' + file_parts [ - <NUM_LIT:1> ] . lstrip ( '<STR_LIT:.>' ) + '<STR_LIT>' , ) ) <EOL> def get_backup_file ( filename ) : <EOL> file_parts = os . path . split ( filename ) <EOL> return os . path . join ( * file_parts [ : - <NUM_LIT:1> ] + ( '<STR_LIT:.>' + file_parts [ - <NUM_LIT:1> ] . lstrip ( '<STR_LIT:.>' ) + '<STR_LIT>' , ) ) <EOL> @ contextmanager <EOL> def atomic_replace ( filename ) : <EOL> """<STR_LIT>""" <EOL> tmpfile_name = get_tmp_file ( filename ) <EOL> fd = os . open ( tmpfile_name , os . O_CREAT | os . O_EXCL | os . O_RDWR , <NUM_LIT> ) <EOL> try : <EOL> f = os . fdopen ( fd , "<STR_LIT>" ) <EOL> yield f <EOL> f . flush ( ) <EOL> os . fsync ( fd ) <EOL> f . seek ( <NUM_LIT:0> ) <EOL> new_content = f . read ( ) <EOL> if not new_content : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> try : <EOL> with open ( filename , '<STR_LIT:rb>' ) as current_f : <EOL> current_content = current_f . read ( ) <EOL> except IOError : <EOL> current_content = b'<STR_LIT>' <EOL> if current_content != new_content : <EOL> with open ( get_backup_file ( filename ) , '<STR_LIT>' ) as backup_file : <EOL> backup_file . write ( current_content ) <EOL> except : <EOL> os . unlink ( tmpfile_name ) <EOL> raise <EOL> else : <EOL> os . rename ( tmpfile_name , filename ) <EOL> finally : <EOL> f . close ( ) <EOL> def edit_in_editor ( current ) : <EOL> EDITOR = os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> with tempfile . NamedTemporaryFile ( mode = '<STR_LIT>' ) as f : <EOL> try : <EOL> f . write ( current ) <EOL> f . flush ( ) <EOL> subprocess . call ( [ EDITOR , f . name ] ) <EOL> f . seek ( <NUM_LIT:0> ) <EOL> return f . read ( ) <EOL> finally : <EOL> f . write ( '<STR_LIT:0>' * os . path . getsize ( f . name ) ) <EOL> f . flush ( ) <EOL> def pretty_record ( record ) : <EOL> s = '<STR_LIT>' % ( record [ <NUM_LIT:1> ] , record [ <NUM_LIT:0> ] ) <EOL> if record [ <NUM_LIT:3> ] : <EOL> s += '<STR_LIT>' + record [ <NUM_LIT:3> ] <EOL> return s <EOL> class InteractiveSession ( object ) : <EOL> def __init__ ( self , args , output = sys . stdout , input = sys . stdin , password = None ) : <EOL> self . args = args <EOL> self . file = args . file <EOL> self . output = output <EOL> self . input = input <EOL> try : <EOL> self . gpg_agent = gpg_agent . GpgAgent ( ) <EOL> except KeyError : <EOL> self . gpg_agent = None <EOL> self . gpg_agent_password_id = '<STR_LIT>' . format ( <EOL> file_fingerprint = hashlib . md5 ( force_bytes ( self . file ) ) . hexdigest ( ) <EOL> ) <EOL> self . password = password <EOL> if not self . password : <EOL> self . password = self . get_master_password ( ) <EOL> def get_master_password ( self , error = None ) : <EOL> if self . password : <EOL> return self . password <EOL> if self . input == sys . stdin : <EOL> if self . gpg_agent : <EOL> error = error or '<STR_LIT:X>' <EOL> self . password = self . gpg_agent . get_passphrase ( <EOL> self . gpg_agent_password_id , <EOL> prompt = '<STR_LIT>' , <EOL> error = error <EOL> ) <EOL> else : <EOL> if error : <EOL> self . output . write ( '<STR_LIT>' . format ( error = error ) ) <EOL> self . password = getpass ( ) <EOL> else : <EOL> self . output . write ( '<STR_LIT>' ) <EOL> self . output . flush ( ) <EOL> self . password = self . input . readline ( ) . rstrip ( '<STR_LIT:\n>' ) <EOL> return self . password <EOL> def clear_master_password ( self ) : <EOL> self . password = None <EOL> if self . gpg_agent : <EOL> self . gpg_agent . clear_passphrase ( self . gpg_agent_password_id ) <EOL> def prompt ( self , prompt = '<STR_LIT>' , required = True , password = False ) : <EOL> while True : <EOL> if password and self . input == sys . stdin : <EOL> line = getpass ( prompt ) <EOL> else : <EOL> self . output . write ( prompt ) <EOL> self . output . flush ( ) <EOL> line = self . input . readline ( ) . rstrip ( '<STR_LIT:\n>' ) <EOL> if not required or line : <EOL> return line <EOL> def get_record ( self , domain = None ) : <EOL> domain = domain or self . prompt ( '<STR_LIT>' ) <EOL> username = self . prompt ( '<STR_LIT>' ) <EOL> password = self . prompt ( <EOL> '<STR_LIT>' , <EOL> required = False , <EOL> password = True <EOL> ) or gen_password_entropy ( <NUM_LIT> ) <EOL> notes = self . prompt ( '<STR_LIT>' , required = False ) <EOL> return ( domain , username , password , notes ) <EOL> def edit_record ( self , record ) : <EOL> new_record = list ( record ) <EOL> new_record [ <NUM_LIT:0> ] = self . prompt ( '<STR_LIT>' % record [ <NUM_LIT:0> ] , required = False ) or record [ <NUM_LIT:0> ] <EOL> new_record [ <NUM_LIT:1> ] = self . prompt ( '<STR_LIT>' % record [ <NUM_LIT:1> ] , required = False ) or record [ <NUM_LIT:1> ] <EOL> pw = self . prompt ( '<STR_LIT>' , required = False , password = True ) or record [ <NUM_LIT:2> ] <EOL> if pw == '<STR_LIT:g>' : <EOL> new_record [ <NUM_LIT:2> ] = gen_password_entropy ( <NUM_LIT> ) <EOL> elif pw : <EOL> new_record [ <NUM_LIT:2> ] = pw <EOL> self . output . write ( "<STR_LIT>" % record [ <NUM_LIT:3> ] ) <EOL> edit = self . prompt ( '<STR_LIT>' , required = False ) or '<STR_LIT:n>' <EOL> if edit [ <NUM_LIT:0> ] == '<STR_LIT:y>' : <EOL> new_record [ <NUM_LIT:3> ] = edit_in_editor ( record [ <NUM_LIT:3> ] ) <EOL> return tuple ( new_record ) <EOL> def find_record ( self , query , records ) : <EOL> possibilities = search ( query , records ) <EOL> if len ( possibilities ) > <NUM_LIT:1> : <EOL> choices = disambiguate ( possibilities ) <EOL> for i , choice in enumerate ( choices ) : <EOL> self . output . write ( '<STR_LIT>' % ( i , choice ) ) <EOL> choice = self . prompt ( '<STR_LIT>' , required = False ) or <NUM_LIT:0> <EOL> return possibilities [ int ( choice ) ] <EOL> else : <EOL> return possibilities [ <NUM_LIT:0> ] <EOL> def read_records ( self , error = None ) : <EOL> try : <EOL> with open ( self . file , '<STR_LIT:rb>' ) as f : <EOL> password = self . get_master_password ( error ) <EOL> try : <EOL> return decode ( decrypt ( password , f . read ( ) ) ) <EOL> except IncorrectPasswordException : <EOL> self . clear_master_password ( ) <EOL> return self . read_records ( error = '<STR_LIT>' ) <EOL> except : <EOL> self . clear_master_password ( ) <EOL> raise <EOL> except IOError : <EOL> return [ ] <EOL> def add_action ( self ) : <EOL> record = self . get_record ( self . args . domain or self . prompt ( '<STR_LIT>' ) ) <EOL> def add ( records ) : <EOL> return records + [ record ] <EOL> self . edit_transaction ( add ) <EOL> def show_action ( self , clipboard = <NUM_LIT:10> ) : <EOL> record = self . find_record ( self . args . domain or self . prompt ( "<STR_LIT>" ) , self . read_records ( ) ) <EOL> self . output . write ( pretty_record ( record ) ) <EOL> self . output . write ( "<STR_LIT:\n>" ) <EOL> if clipboard : <EOL> try : <EOL> self . output . write ( "<STR_LIT>" ) <EOL> set_clipboard_once ( record [ <NUM_LIT:1> ] ) <EOL> self . output . write ( "<STR_LIT>" ) <EOL> set_clipboard_once ( record [ <NUM_LIT:2> ] ) <EOL> except ClipboardException as e : <EOL> self . output . write ( "<STR_LIT>" % e . output . split ( '<STR_LIT:\n>' ) [ <NUM_LIT:0> ] ) <EOL> self . output . write ( record [ <NUM_LIT:2> ] ) <EOL> self . output . write ( "<STR_LIT:\n>" ) <EOL> else : <EOL> return record [ <NUM_LIT:2> ] <EOL> def edit_transaction ( self , callback ) : <EOL> with atomic_replace ( self . file ) as out : <EOL> records = callback ( self . read_records ( ) ) <EOL> assert isinstance ( records , list ) <EOL> if not is_unique_list ( records ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> out . write ( encrypt ( self . password , encode ( records ) ) ) <EOL> out . seek ( <NUM_LIT:0> ) <EOL> assert records == decode ( decrypt ( self . password , out . read ( ) ) ) <EOL> def edit_action ( self ) : <EOL> def edit ( records ) : <EOL> record = self . find_record ( self . args . domain or self . prompt ( '<STR_LIT>' ) , records ) <EOL> new_record = self . edit_record ( record ) <EOL> for i , choice in enumerate ( records ) : <EOL> if choice == record : <EOL> records [ i ] = tuple ( new_record ) <EOL> return records <EOL> self . edit_transaction ( edit ) <EOL> def delete_action ( self ) : <EOL> def delete ( records ) : <EOL> record = self . find_record ( self . args . domain or self . prompt ( '<STR_LIT>' ) , records ) <EOL> self . output . write ( pretty_record ( record ) ) <EOL> self . output . write ( '<STR_LIT:\n>' ) <EOL> confirm = self . prompt ( '<STR_LIT>' , required = False ) or '<STR_LIT:n>' <EOL> if confirm [ <NUM_LIT:0> ] == '<STR_LIT:y>' : <EOL> for i , choice in enumerate ( records ) : <EOL> if choice == record : <EOL> del records [ i ] <EOL> else : <EOL> self . output . write ( "<STR_LIT>" ) <EOL> return records <EOL> self . edit_transaction ( delete ) <EOL> def raw_action ( self ) : <EOL> try : <EOL> output = self . output . buffer <EOL> except AttributeError : <EOL> output = self . output <EOL> output . write ( encode ( self . read_records ( ) ) ) </s>
