pyassuan/common.py

   1 # Copyright
   2
   3 """Items common to both the client and server
   4 """
   5
   6 import re as _re
   7
   8 from . import error as _error
   9
  10
  11 LINE_LENGTH = 1002  # 1000 + [CR,]LF
  12 _ENCODE_REGEXP = _re.compile(
  13     '(' + '|'.join(['%', '\r', '\n']) + ')')
  14 _DECODE_REGEXP = _re.compile('(%[0-9A-F]{2})')
  15 _REQUEST_REGEXP = _re.compile('^(\w+)( *)(.*)\Z')
  16
  17
  18 def encode(string):
  19     r"""
  20
  21     >>> encode('It grew by 5%!\n')
  22     'It grew by 5%25!%0A'
  23     """
  24     return _ENCODE_REGEXP.sub(
  25         lambda x : to_hex(x.group()), string)
  26
  27 def decode(string):
  28     r"""
  29
  30     >>> decode('%22Look out!%22%0AWhere%3F')
  31     '"Look out!"\nWhere?'
  32     """
  33     return _DECODE_REGEXP.sub(
  34         lambda x : from_hex(x.group()), string)
  35
  36 def from_hex(code):
  37     r"""
  38
  39     >>> from_hex('%22')
  40     '"'
  41     >>> from_hex('%0A')
  42     '\n'
  43     """
  44     return chr(int(code[1:], 16))
  45
  46 def to_hex(char):
  47     r"""
  48
  49     >>> to_hex('"')
  50     '%22'
  51     >>> to_hex('\n')
  52     '%0A'
  53     """
  54     return '%{:02X}'.format(ord(char))
  55
  56
  57 class Request (object):
  58     """A client request
  59
  60     http://www.gnupg.org/documentation/manuals/assuan/Client-requests.html
  61
  62     >>> r = Request(command='BYE')
  63     >>> str(r)
  64     'BYE'
  65     >>> r = Request(command='OPTION', parameters='testing at 5%')
  66     >>> str(r)
  67     'OPTION testing at 5%25'
  68     >>> r.from_string('BYE')
  69     >>> r.command
  70     'BYE'
  71     >>> print(r.parameters)
  72     None
  73     >>> r.from_string('OPTION testing at 5%25')
  74     >>> r.command
  75     'OPTION'
  76     >>> print(r.parameters)
  77     testing at 5%
  78     >>> r.from_string(' invalid')
  79     Traceback (most recent call last):
  80       ...
  81     pyassuan.error.AssuanError: 170 Invalid request
  82     >>> r.from_string('in-valid')
  83     Traceback (most recent call last):
  84       ...
  85     pyassuan.error.AssuanError: 170 Invalid request
  86     """
  87     def __init__(self, command=None, parameters=None, encoded=False):
  88         self.command = command
  89         self.parameters = parameters
  90         self.encoded = encoded
  91
  92     def __str__(self):
  93         if self.parameters:
  94             if self.encoded:
  95                 encoded_parameters = self.parameters
  96             else:
  97                 encoded_parameters = encode(self.parameters)
  98             return '{} {}'.format(self.command, encoded_parameters)
  99         return self.command
 100
 101     def from_string(self, string):
 102         if len(string) > 1000:  # TODO: byte-vs-str and newlines?
 103             raise _error.AssuanError(message='Line too long')
 104         match = _REQUEST_REGEXP.match(string)
 105         if not match:
 106             raise _error.AssuanError(message='Invalid request')
 107         self.command = match.group(1)
 108         if match.group(3):
 109             if match.group(2):
 110                 self.parameters = decode(match.group(3))
 111             else:
 112                 raise _error.AssuanError(message='Invalid request')
 113         else:
 114             self.parameters = None
 115
 116
 117 class Response (object):
 118     """A server response
 119
 120     http://www.gnupg.org/documentation/manuals/assuan/Server-responses.html
 121
 122     >>> r = Response(type='OK')
 123     >>> str(r)
 124     'OK'
 125     >>> r = Response(type='ERR', parameters='1 General error')
 126     >>> str(r)
 127     'ERR 1 General error'
 128     >>> r.from_string('OK')
 129     >>> r.type
 130     'OK'
 131     >>> print(r.parameters)
 132     None
 133     >>> r.from_string('ERR 1 General error')
 134     >>> r.type
 135     'ERR'
 136     >>> print(r.parameters)
 137     1 General error
 138     >>> r.from_string(' invalid')
 139     Traceback (most recent call last):
 140       ...
 141     pyassuan.error.AssuanError: 76 Invalid response
 142     >>> r.from_string('in-valid')
 143     Traceback (most recent call last):
 144       ...
 145     pyassuan.error.AssuanError: 76 Invalid response
 146     """
 147     types = {
 148         'O': 'OK',
 149         'E': 'ERR',
 150         'S': 'S',
 151         '#': '#',
 152         'D': 'D',
 153         'I': 'INQUIRE',
 154         }
 155
 156     def __init__(self, type=None, parameters=None):
 157         self.type = type
 158         self.parameters = parameters
 159
 160     def __str__(self):
 161         if self.parameters:
 162             return '{} {}'.format(self.type, encode(self.parameters))
 163         return self.type
 164
 165     def from_string(self, string):
 166         if len(string) > 1000:  # TODO: byte-vs-str and newlines?
 167             raise _error.AssuanError(message='Line too long')
 168         try:
 169             type = self.types[string[0]]
 170         except KeyError:
 171             raise _error.AssuanError(message='Invalid response')
 172         self.type = type
 173         if type == 'D':  # data
 174             self.parameters = decode(string[2:])
 175         elif type == '#':  # comment
 176             self.parameters = decode(string[2:])
 177         else:
 178             match = _REQUEST_REGEXP.match(string)
 179             if not match:
 180                 raise _error.AssuanError(message='Invalid request')
 181             if match.group(3):
 182                 if match.group(2):
 183                     self.parameters = decode(match.group(3))
 184                 else:
 185                     raise _error.AssuanError(message='Invalid request')
 186             else:
 187                 self.parameters = None
 188
 189
 190 def error_response(error):
 191     """
 192
 193     >>> from pyassuan.error import AssuanError
 194     >>> error = AssuanError(1)
 195     >>> response = error_response(error)
 196     >>> print(response)
 197     ERR 1 General error
 198     """
 199     return Response(type='ERR', parameters=str(error))