Merge pull request #2300 from acsbendi/master
Basic IAM authentication, authorization
This commit is contained in:
Steve Pulec
GitHub
commit
69d86cbd54
14 changed files with 1460 additions and 50 deletions
|
|
@ -1,4 +1,7 @@
|
|||
from __future__ import unicode_literals
|
||||
|
||||
from .models import BaseModel, BaseBackend, moto_api_backend # flake8: noqa
|
||||
from .responses import ActionAuthenticatorMixin
|
||||
|
||||
moto_api_backends = {"global": moto_api_backend}
|
||||
set_initial_no_auth_action_count = ActionAuthenticatorMixin.set_initial_no_auth_action_count
|
||||
|
|
|
|||
363
moto/core/access_control.py
Normal file
363
moto/core/access_control.py
Normal file
|
|
@ -0,0 +1,363 @@
|
|||
"""
|
||||
This implementation is NOT complete, there are many things to improve.
|
||||
The following is a list of the most important missing features and inaccuracies.
|
||||
|
||||
TODO add support for more principals, apart from IAM users and assumed IAM roles
|
||||
TODO add support for the Resource and Condition parts of IAM policies
|
||||
TODO add support and create tests for all services in moto (for example, API Gateway is probably not supported currently)
|
||||
TODO implement service specific error messages (currently, EC2 and S3 are supported separately, everything else defaults to the errors IAM returns)
|
||||
TODO include information about the action's resource in error messages (once the Resource element in IAM policies is supported)
|
||||
TODO check all other actions that are performed by the action called by the user (for example, autoscaling:CreateAutoScalingGroup requires permission for iam:CreateServiceLinkedRole too - see https://docs.aws.amazon.com/autoscaling/ec2/userguide/control-access-using-iam.html)
|
||||
TODO add support for resource-based policies
|
||||
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from abc import abstractmethod, ABCMeta
|
||||
from enum import Enum
|
||||
|
||||
import six
|
||||
from botocore.auth import SigV4Auth, S3SigV4Auth
|
||||
from botocore.awsrequest import AWSRequest
|
||||
from botocore.credentials import Credentials
|
||||
from six import string_types
|
||||
|
||||
from moto.iam.models import ACCOUNT_ID, Policy
|
||||
from moto.iam import iam_backend
|
||||
from moto.core.exceptions import SignatureDoesNotMatchError, AccessDeniedError, InvalidClientTokenIdError, AuthFailureError
|
||||
from moto.s3.exceptions import (
|
||||
BucketAccessDeniedError,
|
||||
S3AccessDeniedError,
|
||||
BucketInvalidTokenError,
|
||||
S3InvalidTokenError,
|
||||
S3InvalidAccessKeyIdError,
|
||||
BucketInvalidAccessKeyIdError,
|
||||
BucketSignatureDoesNotMatchError,
|
||||
S3SignatureDoesNotMatchError
|
||||
)
|
||||
from moto.sts import sts_backend
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def create_access_key(access_key_id, headers):
|
||||
if access_key_id.startswith("AKIA") or "X-Amz-Security-Token" not in headers:
|
||||
return IAMUserAccessKey(access_key_id, headers)
|
||||
else:
|
||||
return AssumedRoleAccessKey(access_key_id, headers)
|
||||
|
||||
|
||||
class IAMUserAccessKey(object):
|
||||
|
||||
def __init__(self, access_key_id, headers):
|
||||
iam_users = iam_backend.list_users('/', None, None)
|
||||
for iam_user in iam_users:
|
||||
for access_key in iam_user.access_keys:
|
||||
if access_key.access_key_id == access_key_id:
|
||||
self._owner_user_name = iam_user.name
|
||||
self._access_key_id = access_key_id
|
||||
self._secret_access_key = access_key.secret_access_key
|
||||
if "X-Amz-Security-Token" in headers:
|
||||
raise CreateAccessKeyFailure(reason="InvalidToken")
|
||||
return
|
||||
raise CreateAccessKeyFailure(reason="InvalidId")
|
||||
|
||||
@property
|
||||
def arn(self):
|
||||
return "arn:aws:iam::{account_id}:user/{iam_user_name}".format(
|
||||
account_id=ACCOUNT_ID,
|
||||
iam_user_name=self._owner_user_name
|
||||
)
|
||||
|
||||
def create_credentials(self):
|
||||
return Credentials(self._access_key_id, self._secret_access_key)
|
||||
|
||||
def collect_policies(self):
|
||||
user_policies = []
|
||||
|
||||
inline_policy_names = iam_backend.list_user_policies(self._owner_user_name)
|
||||
for inline_policy_name in inline_policy_names:
|
||||
inline_policy = iam_backend.get_user_policy(self._owner_user_name, inline_policy_name)
|
||||
user_policies.append(inline_policy)
|
||||
|
||||
attached_policies, _ = iam_backend.list_attached_user_policies(self._owner_user_name)
|
||||
user_policies += attached_policies
|
||||
|
||||
user_groups = iam_backend.get_groups_for_user(self._owner_user_name)
|
||||
for user_group in user_groups:
|
||||
inline_group_policy_names = iam_backend.list_group_policies(user_group.name)
|
||||
for inline_group_policy_name in inline_group_policy_names:
|
||||
inline_user_group_policy = iam_backend.get_group_policy(user_group.name, inline_group_policy_name)
|
||||
user_policies.append(inline_user_group_policy)
|
||||
|
||||
attached_group_policies, _ = iam_backend.list_attached_group_policies(user_group.name)
|
||||
user_policies += attached_group_policies
|
||||
|
||||
return user_policies
|
||||
|
||||
|
||||
class AssumedRoleAccessKey(object):
|
||||
|
||||
def __init__(self, access_key_id, headers):
|
||||
for assumed_role in sts_backend.assumed_roles:
|
||||
if assumed_role.access_key_id == access_key_id:
|
||||
self._access_key_id = access_key_id
|
||||
self._secret_access_key = assumed_role.secret_access_key
|
||||
self._session_token = assumed_role.session_token
|
||||
self._owner_role_name = assumed_role.arn.split("/")[-1]
|
||||
self._session_name = assumed_role.session_name
|
||||
if headers["X-Amz-Security-Token"] != self._session_token:
|
||||
raise CreateAccessKeyFailure(reason="InvalidToken")
|
||||
return
|
||||
raise CreateAccessKeyFailure(reason="InvalidId")
|
||||
|
||||
@property
|
||||
def arn(self):
|
||||
return "arn:aws:sts::{account_id}:assumed-role/{role_name}/{session_name}".format(
|
||||
account_id=ACCOUNT_ID,
|
||||
role_name=self._owner_role_name,
|
||||
session_name=self._session_name
|
||||
)
|
||||
|
||||
def create_credentials(self):
|
||||
return Credentials(self._access_key_id, self._secret_access_key, self._session_token)
|
||||
|
||||
def collect_policies(self):
|
||||
role_policies = []
|
||||
|
||||
inline_policy_names = iam_backend.list_role_policies(self._owner_role_name)
|
||||
for inline_policy_name in inline_policy_names:
|
||||
_, inline_policy = iam_backend.get_role_policy(self._owner_role_name, inline_policy_name)
|
||||
role_policies.append(inline_policy)
|
||||
|
||||
attached_policies, _ = iam_backend.list_attached_role_policies(self._owner_role_name)
|
||||
role_policies += attached_policies
|
||||
|
||||
return role_policies
|
||||
|
||||
|
||||
class CreateAccessKeyFailure(Exception):
|
||||
|
||||
def __init__(self, reason, *args):
|
||||
super(CreateAccessKeyFailure, self).__init__(*args)
|
||||
self.reason = reason
|
||||
|
||||
|
||||
@six.add_metaclass(ABCMeta)
|
||||
class IAMRequestBase(object):
|
||||
|
||||
def __init__(self, method, path, data, headers):
|
||||
log.debug("Creating {class_name} with method={method}, path={path}, data={data}, headers={headers}".format(
|
||||
class_name=self.__class__.__name__, method=method, path=path, data=data, headers=headers))
|
||||
self._method = method
|
||||
self._path = path
|
||||
self._data = data
|
||||
self._headers = headers
|
||||
credential_scope = self._get_string_between('Credential=', ',', self._headers['Authorization'])
|
||||
credential_data = credential_scope.split('/')
|
||||
self._region = credential_data[2]
|
||||
self._service = credential_data[3]
|
||||
self._action = self._service + ":" + (self._data["Action"][0] if isinstance(self._data["Action"], list) else self._data["Action"])
|
||||
try:
|
||||
self._access_key = create_access_key(access_key_id=credential_data[0], headers=headers)
|
||||
except CreateAccessKeyFailure as e:
|
||||
self._raise_invalid_access_key(e.reason)
|
||||
|
||||
def check_signature(self):
|
||||
original_signature = self._get_string_between('Signature=', ',', self._headers['Authorization'])
|
||||
calculated_signature = self._calculate_signature()
|
||||
if original_signature != calculated_signature:
|
||||
self._raise_signature_does_not_match()
|
||||
|
||||
def check_action_permitted(self):
|
||||
policies = self._access_key.collect_policies()
|
||||
|
||||
permitted = False
|
||||
for policy in policies:
|
||||
iam_policy = IAMPolicy(policy)
|
||||
permission_result = iam_policy.is_action_permitted(self._action)
|
||||
if permission_result == PermissionResult.DENIED:
|
||||
self._raise_access_denied()
|
||||
elif permission_result == PermissionResult.PERMITTED:
|
||||
permitted = True
|
||||
|
||||
if not permitted:
|
||||
self._raise_access_denied()
|
||||
|
||||
@abstractmethod
|
||||
def _raise_signature_does_not_match(self):
|
||||
raise NotImplementedError()
|
||||
|
||||
@abstractmethod
|
||||
def _raise_access_denied(self):
|
||||
raise NotImplementedError()
|
||||
|
||||
@abstractmethod
|
||||
def _raise_invalid_access_key(self, reason):
|
||||
raise NotImplementedError()
|
||||
|
||||
@abstractmethod
|
||||
def _create_auth(self, credentials):
|
||||
raise NotImplementedError()
|
||||
|
||||
@staticmethod
|
||||
def _create_headers_for_aws_request(signed_headers, original_headers):
|
||||
headers = {}
|
||||
for key, value in original_headers.items():
|
||||
if key.lower() in signed_headers:
|
||||
headers[key] = value
|
||||
return headers
|
||||
|
||||
def _create_aws_request(self):
|
||||
signed_headers = self._get_string_between('SignedHeaders=', ',', self._headers['Authorization']).split(';')
|
||||
headers = self._create_headers_for_aws_request(signed_headers, self._headers)
|
||||
request = AWSRequest(method=self._method, url=self._path, data=self._data, headers=headers)
|
||||
request.context['timestamp'] = headers['X-Amz-Date']
|
||||
|
||||
return request
|
||||
|
||||
def _calculate_signature(self):
|
||||
credentials = self._access_key.create_credentials()
|
||||
auth = self._create_auth(credentials)
|
||||
request = self._create_aws_request()
|
||||
canonical_request = auth.canonical_request(request)
|
||||
string_to_sign = auth.string_to_sign(request, canonical_request)
|
||||
return auth.signature(string_to_sign, request)
|
||||
|
||||
@staticmethod
|
||||
def _get_string_between(first_separator, second_separator, string):
|
||||
return string.partition(first_separator)[2].partition(second_separator)[0]
|
||||
|
||||
|
||||
class IAMRequest(IAMRequestBase):
|
||||
|
||||
def _raise_signature_does_not_match(self):
|
||||
if self._service == "ec2":
|
||||
raise AuthFailureError()
|
||||
else:
|
||||
raise SignatureDoesNotMatchError()
|
||||
|
||||
def _raise_invalid_access_key(self, _):
|
||||
if self._service == "ec2":
|
||||
raise AuthFailureError()
|
||||
else:
|
||||
raise InvalidClientTokenIdError()
|
||||
|
||||
def _create_auth(self, credentials):
|
||||
return SigV4Auth(credentials, self._service, self._region)
|
||||
|
||||
def _raise_access_denied(self):
|
||||
raise AccessDeniedError(
|
||||
user_arn=self._access_key.arn,
|
||||
action=self._action
|
||||
)
|
||||
|
||||
|
||||
class S3IAMRequest(IAMRequestBase):
|
||||
|
||||
def _raise_signature_does_not_match(self):
|
||||
if "BucketName" in self._data:
|
||||
raise BucketSignatureDoesNotMatchError(bucket=self._data["BucketName"])
|
||||
else:
|
||||
raise S3SignatureDoesNotMatchError()
|
||||
|
||||
def _raise_invalid_access_key(self, reason):
|
||||
if reason == "InvalidToken":
|
||||
if "BucketName" in self._data:
|
||||
raise BucketInvalidTokenError(bucket=self._data["BucketName"])
|
||||
else:
|
||||
raise S3InvalidTokenError()
|
||||
else:
|
||||
if "BucketName" in self._data:
|
||||
raise BucketInvalidAccessKeyIdError(bucket=self._data["BucketName"])
|
||||
else:
|
||||
raise S3InvalidAccessKeyIdError()
|
||||
|
||||
def _create_auth(self, credentials):
|
||||
return S3SigV4Auth(credentials, self._service, self._region)
|
||||
|
||||
def _raise_access_denied(self):
|
||||
if "BucketName" in self._data:
|
||||
raise BucketAccessDeniedError(bucket=self._data["BucketName"])
|
||||
else:
|
||||
raise S3AccessDeniedError()
|
||||
|
||||
|
||||
class IAMPolicy(object):
|
||||
|
||||
def __init__(self, policy):
|
||||
if isinstance(policy, Policy):
|
||||
default_version = next(policy_version for policy_version in policy.versions if policy_version.is_default)
|
||||
policy_document = default_version.document
|
||||
elif isinstance(policy, string_types):
|
||||
policy_document = policy
|
||||
else:
|
||||
policy_document = policy["policy_document"]
|
||||
|
||||
self._policy_json = json.loads(policy_document)
|
||||
|
||||
def is_action_permitted(self, action):
|
||||
permitted = False
|
||||
if isinstance(self._policy_json["Statement"], list):
|
||||
for policy_statement in self._policy_json["Statement"]:
|
||||
iam_policy_statement = IAMPolicyStatement(policy_statement)
|
||||
permission_result = iam_policy_statement.is_action_permitted(action)
|
||||
if permission_result == PermissionResult.DENIED:
|
||||
return permission_result
|
||||
elif permission_result == PermissionResult.PERMITTED:
|
||||
permitted = True
|
||||
else: # dict
|
||||
iam_policy_statement = IAMPolicyStatement(self._policy_json["Statement"])
|
||||
return iam_policy_statement.is_action_permitted(action)
|
||||
|
||||
if permitted:
|
||||
return PermissionResult.PERMITTED
|
||||
else:
|
||||
return PermissionResult.NEUTRAL
|
||||
|
||||
|
||||
class IAMPolicyStatement(object):
|
||||
|
||||
def __init__(self, statement):
|
||||
self._statement = statement
|
||||
|
||||
def is_action_permitted(self, action):
|
||||
is_action_concerned = False
|
||||
|
||||
if "NotAction" in self._statement:
|
||||
if not self._check_element_matches("NotAction", action):
|
||||
is_action_concerned = True
|
||||
else: # Action is present
|
||||
if self._check_element_matches("Action", action):
|
||||
is_action_concerned = True
|
||||
|
||||
if is_action_concerned:
|
||||
if self._statement["Effect"] == "Allow":
|
||||
return PermissionResult.PERMITTED
|
||||
else: # Deny
|
||||
return PermissionResult.DENIED
|
||||
else:
|
||||
return PermissionResult.NEUTRAL
|
||||
|
||||
def _check_element_matches(self, statement_element, value):
|
||||
if isinstance(self._statement[statement_element], list):
|
||||
for statement_element_value in self._statement[statement_element]:
|
||||
if self._match(statement_element_value, value):
|
||||
return True
|
||||
return False
|
||||
else: # string
|
||||
return self._match(self._statement[statement_element], value)
|
||||
|
||||
@staticmethod
|
||||
def _match(pattern, string):
|
||||
pattern = pattern.replace("*", ".*")
|
||||
pattern = "^{pattern}$".format(pattern=pattern)
|
||||
return re.match(pattern, string)
|
||||
|
||||
|
||||
class PermissionResult(Enum):
|
||||
PERMITTED = 1
|
||||
DENIED = 2
|
||||
NEUTRAL = 3
|
||||
|
|
@ -65,3 +65,42 @@ class JsonRESTError(RESTError):
|
|||
|
||||
def get_body(self, *args, **kwargs):
|
||||
return self.description
|
||||
|
||||
|
||||
class SignatureDoesNotMatchError(RESTError):
|
||||
code = 403
|
||||
|
||||
def __init__(self):
|
||||
super(SignatureDoesNotMatchError, self).__init__(
|
||||
'SignatureDoesNotMatch',
|
||||
"The request signature we calculated does not match the signature you provided. Check your AWS Secret Access Key and signing method. Consult the service documentation for details.")
|
||||
|
||||
|
||||
class InvalidClientTokenIdError(RESTError):
|
||||
code = 403
|
||||
|
||||
def __init__(self):
|
||||
super(InvalidClientTokenIdError, self).__init__(
|
||||
'InvalidClientTokenId',
|
||||
"The security token included in the request is invalid.")
|
||||
|
||||
|
||||
class AccessDeniedError(RESTError):
|
||||
code = 403
|
||||
|
||||
def __init__(self, user_arn, action):
|
||||
super(AccessDeniedError, self).__init__(
|
||||
'AccessDenied',
|
||||
"User: {user_arn} is not authorized to perform: {operation}".format(
|
||||
user_arn=user_arn,
|
||||
operation=action
|
||||
))
|
||||
|
||||
|
||||
class AuthFailureError(RESTError):
|
||||
code = 401
|
||||
|
||||
def __init__(self):
|
||||
super(AuthFailureError, self).__init__(
|
||||
'AuthFailure',
|
||||
"AWS was not able to validate the provided access credentials")
|
||||
|
|
|
|||
|
|
@ -1,13 +1,17 @@
|
|||
from __future__ import unicode_literals
|
||||
|
||||
import functools
|
||||
from collections import defaultdict
|
||||
import datetime
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import io
|
||||
import requests
|
||||
|
||||
import pytz
|
||||
|
||||
from moto.core.access_control import IAMRequest, S3IAMRequest
|
||||
from moto.core.exceptions import DryRunClientError
|
||||
|
||||
from jinja2 import Environment, DictLoader, TemplateNotFound
|
||||
|
|
@ -22,7 +26,7 @@ from werkzeug.exceptions import HTTPException
|
|||
import boto3
|
||||
from moto.compat import OrderedDict
|
||||
from moto.core.utils import camelcase_to_underscores, method_names_from_class
|
||||
|
||||
from moto import settings
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -103,7 +107,54 @@ class _TemplateEnvironmentMixin(object):
|
|||
return self.environment.get_template(template_id)
|
||||
|
||||
|
||||
class BaseResponse(_TemplateEnvironmentMixin):
|
||||
class ActionAuthenticatorMixin(object):
|
||||
|
||||
request_count = 0
|
||||
|
||||
def _authenticate_and_authorize_action(self, iam_request_cls):
|
||||
if ActionAuthenticatorMixin.request_count >= settings.INITIAL_NO_AUTH_ACTION_COUNT:
|
||||
iam_request = iam_request_cls(method=self.method, path=self.path, data=self.data, headers=self.headers)
|
||||
iam_request.check_signature()
|
||||
iam_request.check_action_permitted()
|
||||
else:
|
||||
ActionAuthenticatorMixin.request_count += 1
|
||||
|
||||
def _authenticate_and_authorize_normal_action(self):
|
||||
self._authenticate_and_authorize_action(IAMRequest)
|
||||
|
||||
def _authenticate_and_authorize_s3_action(self):
|
||||
self._authenticate_and_authorize_action(S3IAMRequest)
|
||||
|
||||
@staticmethod
|
||||
def set_initial_no_auth_action_count(initial_no_auth_action_count):
|
||||
def decorator(function):
|
||||
def wrapper(*args, **kwargs):
|
||||
if settings.TEST_SERVER_MODE:
|
||||
response = requests.post("http://localhost:5000/moto-api/reset-auth", data=str(initial_no_auth_action_count).encode())
|
||||
original_initial_no_auth_action_count = response.json()['PREVIOUS_INITIAL_NO_AUTH_ACTION_COUNT']
|
||||
else:
|
||||
original_initial_no_auth_action_count = settings.INITIAL_NO_AUTH_ACTION_COUNT
|
||||
original_request_count = ActionAuthenticatorMixin.request_count
|
||||
settings.INITIAL_NO_AUTH_ACTION_COUNT = initial_no_auth_action_count
|
||||
ActionAuthenticatorMixin.request_count = 0
|
||||
try:
|
||||
result = function(*args, **kwargs)
|
||||
finally:
|
||||
if settings.TEST_SERVER_MODE:
|
||||
requests.post("http://localhost:5000/moto-api/reset-auth", data=str(original_initial_no_auth_action_count).encode())
|
||||
else:
|
||||
ActionAuthenticatorMixin.request_count = original_request_count
|
||||
settings.INITIAL_NO_AUTH_ACTION_COUNT = original_initial_no_auth_action_count
|
||||
return result
|
||||
|
||||
functools.update_wrapper(wrapper, function)
|
||||
wrapper.__wrapped__ = function
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
|
||||
|
||||
default_region = 'us-east-1'
|
||||
# to extract region, use [^.]
|
||||
|
|
@ -167,6 +218,7 @@ class BaseResponse(_TemplateEnvironmentMixin):
|
|||
self.uri = full_url
|
||||
self.path = urlparse(full_url).path
|
||||
self.querystring = querystring
|
||||
self.data = querystring
|
||||
self.method = request.method
|
||||
self.region = self.get_region_from_url(request, full_url)
|
||||
self.uri_match = None
|
||||
|
|
@ -273,6 +325,13 @@ class BaseResponse(_TemplateEnvironmentMixin):
|
|||
|
||||
def call_action(self):
|
||||
headers = self.response_headers
|
||||
|
||||
try:
|
||||
self._authenticate_and_authorize_normal_action()
|
||||
except HTTPException as http_error:
|
||||
response = http_error.description, dict(status=http_error.code)
|
||||
return self._send_response(headers, response)
|
||||
|
||||
action = camelcase_to_underscores(self._get_action())
|
||||
method_names = method_names_from_class(self.__class__)
|
||||
if action in method_names:
|
||||
|
|
@ -285,16 +344,7 @@ class BaseResponse(_TemplateEnvironmentMixin):
|
|||
if isinstance(response, six.string_types):
|
||||
return 200, headers, response
|
||||
else:
|
||||
if len(response) == 2:
|
||||
body, new_headers = response
|
||||
else:
|
||||
status, new_headers, body = response
|
||||
status = new_headers.get('status', 200)
|
||||
headers.update(new_headers)
|
||||
# Cast status to string
|
||||
if "status" in headers:
|
||||
headers['status'] = str(headers['status'])
|
||||
return status, headers, body
|
||||
return self._send_response(headers, response)
|
||||
|
||||
if not action:
|
||||
return 404, headers, ''
|
||||
|
|
@ -302,6 +352,19 @@ class BaseResponse(_TemplateEnvironmentMixin):
|
|||
raise NotImplementedError(
|
||||
"The {0} action has not been implemented".format(action))
|
||||
|
||||
@staticmethod
|
||||
def _send_response(headers, response):
|
||||
if len(response) == 2:
|
||||
body, new_headers = response
|
||||
else:
|
||||
status, new_headers, body = response
|
||||
status = new_headers.get('status', 200)
|
||||
headers.update(new_headers)
|
||||
# Cast status to string
|
||||
if "status" in headers:
|
||||
headers['status'] = str(headers['status'])
|
||||
return status, headers, body
|
||||
|
||||
def _get_param(self, param_name, if_none=None):
|
||||
val = self.querystring.get(param_name)
|
||||
if val is not None:
|
||||
|
|
@ -569,6 +632,14 @@ class MotoAPIResponse(BaseResponse):
|
|||
return 200, {}, json.dumps({"status": "ok"})
|
||||
return 400, {}, json.dumps({"Error": "Need to POST to reset Moto"})
|
||||
|
||||
def reset_auth_response(self, request, full_url, headers):
|
||||
if request.method == "POST":
|
||||
previous_initial_no_auth_action_count = settings.INITIAL_NO_AUTH_ACTION_COUNT
|
||||
settings.INITIAL_NO_AUTH_ACTION_COUNT = float(request.data.decode())
|
||||
ActionAuthenticatorMixin.request_count = 0
|
||||
return 200, {}, json.dumps({"status": "ok", "PREVIOUS_INITIAL_NO_AUTH_ACTION_COUNT": str(previous_initial_no_auth_action_count)})
|
||||
return 400, {}, json.dumps({"Error": "Need to POST to reset Moto Auth"})
|
||||
|
||||
def model_data(self, request, full_url, headers):
|
||||
from moto.core.models import model_data
|
||||
|
||||
|
|
|
|||
|
|
@ -11,4 +11,5 @@ url_paths = {
|
|||
'{0}/moto-api/$': response_instance.dashboard,
|
||||
'{0}/moto-api/data.json': response_instance.model_data,
|
||||
'{0}/moto-api/reset': response_instance.reset_response,
|
||||
'{0}/moto-api/reset-auth': response_instance.reset_auth_response,
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue