rm: escape_early_upper and internal refactoring

Author: AstinCHOI

README.rst

@@ -106,8 +106,7 @@ Usage
 
  AuthToken(token_type=None, token_name='__token__', key=None, algorithm='sha256', 
  salt=None, start_time=None, end_time=None, window_seconds=None,
- field_delimiter='~', acl_delimiter='!', escape_early=False, 
- escape_early_upper=False, verbose=False)
+ field_delimiter='~', acl_delimiter='!', escape_early=False, verbose=False)
 
 #
 
@@ -125,7 +124,6 @@ Usage
  field_delimiter Character used to delimit token body fields. [Default: ~]
  acl_delimiter Character used to delimit acl fields. [Default: !]
  escape_early Causes strings to be 'url' encoded before being used.
- escape_early_upper Causes strings to be 'url' encoded before being used.
  verbose Print all parameters.
  ==================== ===================================================================================================
 

akamai/authtoken/authtoken.py

@@ -50,7 +50,7 @@ def __init__(self, token_type=None, token_name='__token__',
  key=None, algorithm='sha256', salt=None,
  start_time=None, end_time=None, window_seconds=None,
  field_delimiter='~', acl_delimiter='!',
- escape_early=False, escape_early_upper=False, verbose=False):
+ escape_early=False, verbose=False):
 
  self.token_type = token_type
  self.token_name = token_name
@@ -66,31 +66,27 @@ def __init__(self, token_type=None, token_name='__token__',
  self.field_delimiter = field_delimiter
  self.acl_delimiter = acl_delimiter
  self.escape_early = escape_early
- self.escape_early_upper = escape_early_upper
  self.verbose = verbose
 
- def escapeEarly(self, text):
- if self.escape_early or self.escape_early_upper:
- # Only escape the text if we are configured for escape early.
- new_text = quote_plus(text)
- if self.escape_early_upper:
- def toUpper(match):
- return match.group(1).upper()
- return re.sub(r'(%..)', toUpper, new_text)
- else:
- def toLower(match):
- return match.group(1).lower()
- return re.sub(r'(%..)', toLower, new_text)
-
- # Return the original, unmodified text.
- return text
+ def _escapeEarly(self, text):
+ if self.escape_early:
+ def toLower(match):
+ return match.group(1).lower()
+ return re.sub(r'(%..)', toLower, quote_plus(text))
+ else:
+ return text
 
- def _generateToken(self, url=None, acl=None,
- window_seconds=None, start_time=None, end_time=None,
+ def generateToken(self, url=None, acl=None, start_time=None, 
+ end_time=None, window_seconds=None,
  ip=None, payload=None, session_id=None):
+ if not start_time:
+ start_time = self.start_time
+ if not end_time:
+ end_time = self.end_time
+ if not window_seconds:
+ window_seconds = self.window_seconds
+
  if str(start_time).lower() == 'now':
- # Initialize the start time if we are asked for a starting time of
- # now.
  start_time = int(time.mktime(time.gmtime()))
  elif start_time is not None:
  try:
@@ -113,7 +109,7 @@ def _generateToken(self, url=None, acl=None,
  if end_time is None:
  if int(window_seconds or 0) > 0:
  if start_time is None:
- # If we have a window window without a start time,
+ # If we have a window_seconds without a start time,
  # calculate the end time starting from the current time.
  end_time = int(time.mktime(time.gmtime())) + \
  window_seconds
@@ -123,21 +119,11 @@ def _generateToken(self, url=None, acl=None,
  raise AuthTokenError('You must provide an expiration time or '
  'a duration window..')
 
- try:
- if end_time < start_time:
- raise AuthTokenError('Token will have already expired.')
- except TypeError:
- pass
+ if start_time and (end_time < start_time):
+ raise AuthTokenError('Token will have already expired.')
 
- if ((acl is None and url is None) or
- acl is not None and url is not None and
- (len(acl) <= 0) and (len(url) <= 0)):
- raise AuthTokenError('You must provide a URL or an ACL.')
-
- if (acl is not None and url is not None and
- (len(acl) > 0) and (len(url) > 0)):
- raise AuthTokenError('You must provide a URL OR an ACL, '
- 'not both.')
+ if (not acl and not url) or (acl and url):
+ raise AuthTokenError('You must provide a URL or an ACL')
 
  if self.verbose:
  print('''
@@ -159,77 +145,56 @@ def _generateToken(self, url=None, acl=None,
 ACL Delimiter : {6}
 Escape Early : {7}
 Generating token...'''.format(self.token_type if self.token_type else '', #0
- self.token_name if self.token_name else '', #1
- self.key if self.key else '', #2
- self.algorithm if self.algorithm else '', #3
- self.salt if self.salt else '', #4
- self.field_delimiter if self.field_delimiter else '', #5
- self.acl_delimiter if self.acl_delimiter else '', #6
- self.escape_early if self.escape_early else '', #7
- start_time if start_time else '', #8
- end_time if end_time else '', #9
- window_seconds if window_seconds else '', #10
- ip if ip else '', #11
- url if url else '', #12
- acl if acl else '', #13
- payload if payload else '', #14
- session_id if session_id else '')) #15
-
- hash_source = ''
- new_token = ''
+  self.token_name if self.token_name else '', #1
+  self.key if self.key else '', #2
+  self.algorithm if self.algorithm else '', #3
+  self.salt if self.salt else '', #4
+  self.field_delimiter if self.field_delimiter else '', #5
+  self.acl_delimiter if self.acl_delimiter else '', #6
+  self.escape_early if self.escape_early else '', #7
+  start_time if start_time else '', #8
+  end_time if end_time else '', #9
+  window_seconds if window_seconds else '', #10
+  ip if ip else '', #11
+  url if url else '', #12
+  acl if acl else '', #13
+  payload if payload else '', #14
+  session_id if session_id else '')) #15
+
+ hash_source = []
+ new_token = []
 
  if ip:
- new_token += 'ip={0}{1}'.format(self.escapeEarly(ip),
- self.field_delimiter)
+ new_token.append('ip={0}'.format(self._escapeEarly(ip)))
 
- if start_time is not None:
- new_token += 'st={0}{1}'.format(start_time, self.field_delimiter)
+ if start_time:
+ new_token.append('st={0}'.format(start_time))
 
- new_token += 'exp={0}{1}'.format(end_time, self.field_delimiter)
+ new_token.append('exp={0}'.format(end_time))
 
  if acl:
- new_token += 'acl={0}{1}'.format(acl, self.field_delimiter)
+ new_token.append('acl={0}'.format(acl))
 
  if session_id:
- new_token += 'id={0}{1}'.format(self.escapeEarly(session_id), self.field_delimiter)
+ new_token.append('id={0}'.format(self._escapeEarly(session_id)))
 
  if payload:
- new_token += 'data={0}{1}'.format(self.escapeEarly(payload), self.field_delimiter)
+ new_token.append('data={0}'.format(self._escapeEarly(payload)))
 
- hash_source += new_token
+ hash_source = list(new_token)
  if url and not acl:
- hash_source += 'url={0}{1}'.format(self.escapeEarly(url), self.field_delimiter)
+ hash_source.append('url={0}'.format(self._escapeEarly(url)))
 
  if self.salt:
- hash_source += 'salt={0}{1}'.format(self.salt, self.field_delimiter)
+ hash_source.append('salt={0}'.format(self.salt))
 
  if self.algorithm.lower() not in ('sha256', 'sha1', 'md5'):
  raise AuthTokenError('Unknown algorithm')
 
  token_hmac = hmac.new(
  binascii.a2b_hex(self.key.encode()),
- hash_source.rstrip(self.field_delimiter).encode(),
+ self.field_delimiter.join(hash_source).encode(),
  getattr(hashlib, self.algorithm.lower())).hexdigest()
- new_token += 'hmac={0}'.format(token_hmac)
-
- return new_token
-
- def generateToken(self, url=None, acl=None, start_time=None, end_time=None, window_seconds=None,
- ip=None, payload=None, session_id=None):
- if not start_time:
- start_time = self.start_time
- 
- if not end_time:
- end_time = self.end_time
- 
- if not window_seconds:
- window_seconds = self.window_seconds
+ new_token.append('hmac={0}'.format(token_hmac))
 
- return self._generateToken(url=url,
- acl=acl,
- start_time=start_time,
- end_time=end_time,
- window_seconds=window_seconds,
- ip=ip,
- payload=payload,
- session_id=session_id)
+ return self.field_delimiter.join(new_token)

cms_authtoken.py

@@ -52,10 +52,6 @@
  '-x', '--escape_early',
  action='store_true', default=False, dest='escape_early',
  help='Causes strings to be url encoded before being used.')
- parser.add_option(
- '-X', '--escape_early_upper',
- action='store_true', default=False, dest='escape_early_upper',
- help='Causes strings to be url encoded before being used.')
  parser.add_option(
  '-v', '--verbose',
  action='store_true', default=False, dest='verbose',
@@ -94,7 +90,6 @@
  field_delimiter=options.field_delimiter,
  acl_delimiter=options.acl_delimiter,
  escape_early=options.escape_early,
- escape_early_upper=options.escape_early_upper,
  verbose=options.verbose)
  token = generator.generateToken(url=options.url,
  acl=options.access_list,

test/test_authtoken.py

@@ -31,7 +31,6 @@
 import requests
 
 
-
 if 'TEST_MODE' in os.environ and os.environ['TEST_MODE'] == 'LOCAL':
  # export TEST_MODE=LOCAL
  import secrets
@@ -47,11 +46,12 @@
 
 DEFAULT_WINDOW_SECONDS = 500
 
+
 class TestAuthToken(unittest.TestCase):
 
  def setUp(self):
  # Test for Query String
- self.at = AuthToken(key=AT_ENCRYPTION_KEY, window_seconds=DEFAULT_WINDOW_SECONDS)
+ self.at = AuthToken(**{'key': AT_ENCRYPTION_KEY, 'window_seconds': DEFAULT_WINDOW_SECONDS})
 
  # Test for Cookie
  self.cat = AuthToken(key=AT_ENCRYPTION_KEY, algorithm='sha1', window_seconds=DEFAULT_WINDOW_SECONDS)