Use the registry API for security scanning

when the storage engine doesn't support direct download url

endpoints/v2/v2auth.py

@@ -1,22 +1,20 @@
 import logging
 import re
-import time
-import jwt
 
 from flask import request, jsonify, abort
-from cachetools import lru_cache
 
 from app import app, userevents
 from data import model
 from auth.auth import process_auth
-from auth.registry_jwt_auth import build_context_and_subject
 from auth.auth_context import get_authenticated_user, get_validated_token, get_validated_oauth_token
 from auth.permissions import (ModifyRepositoryPermission, ReadRepositoryPermission,
                               CreateRepositoryPermission)
 from endpoints.v2 import v2_bp
+from endpoints.decorators import anon_protect
 from util.cache import no_cache
 from util.names import parse_namespace_repository, REPOSITORY_NAME_REGEX
-from endpoints.decorators import anon_protect
+from util.security.registry_jwt import generate_jwt_object, build_context_and_subject
+
 
 logger = logging.getLogger(__name__)
 
@@ -26,18 +24,6 @@ SCOPE_REGEX = re.compile(
     r'^repository:(([\.a-zA-Z0-9_\-]+/)?[\.a-zA-Z0-9_\-]+):(((push|pull|\*),)*(push|pull|\*))$'
 )
 
-@lru_cache(maxsize=1)
-def load_certificate_bytes(certificate_file_path):
-  with open(certificate_file_path) as cert_file:
-    return ''.join(cert_file.readlines()[1:-1]).rstrip('\n')
-
-
-@lru_cache(maxsize=1)
-def load_private_key(private_key_file_path):
-  with open(private_key_file_path) as private_key_file:
-    return private_key_file.read()
-
-
 @v2_bp.route('/auth')
 @process_auth
 @no_cache
@@ -156,23 +142,8 @@ def generate_registry_jwt():
 
   # Build the signed JWT.
   context, subject = build_context_and_subject(user, token, oauthtoken)
-  token_data = {
-    'iss': app.config['JWT_AUTH_TOKEN_ISSUER'],
-    'aud': audience_param,
-    'nbf': int(time.time()),
-    'iat': int(time.time()),
-    'exp': int(time.time() + TOKEN_VALIDITY_LIFETIME_S),
-    'sub': subject,
-    'access': access,
-    'context': context,
-  }
 
-  certificate = load_certificate_bytes(app.config['JWT_AUTH_CERTIFICATE_PATH'])
+  jwt_obj = generate_jwt_object(audience_param, subject, context, access, TOKEN_VALIDITY_LIFETIME_S,
+                                app.config)
 
-  token_headers = {
-    'x5c': [certificate],
-  }
-
-  private_key = load_private_key(app.config['JWT_AUTH_PRIVATE_KEY_PATH'])
-
-  return jsonify({'token':jwt.encode(token_data, private_key, 'RS256', headers=token_headers)})
+  return jsonify({'token': jwt_obj})