A more deterministic kms

moto/dynamodbstreams/models.py

@@ -1,13 +1,13 @@
 from __future__ import annotations
 
 import base64
-import os
 from typing import Any, Optional
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
 from moto.dynamodb.models import DynamoDBBackend, dynamodb_backends
 from moto.dynamodb.models.table import StreamShard, Table
+from moto.moto_api._internal import mock_random as random
 
 
 class ShardIterator(BaseModel):
@@ -18,7 +18,7 @@ def __init__(
         shard_iterator_type: str,
         sequence_number: Optional[int] = None,
     ):
-        self.id = base64.b64encode(os.urandom(472)).decode("utf-8")
+        self.id = base64.b64encode(random.randbytes(472)).decode("utf-8")
         self.streams_backend = streams_backend
         self.stream_shard = stream_shard
         self.shard_iterator_type = shard_iterator_type

moto/iam/models.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import json
-import os
 import re
 import string
 from collections.abc import Iterable
@@ -128,7 +127,7 @@ def __init__(self, account_id: str, region_name: str, device_name: str):
             random.choice(string.ascii_uppercase + "234567") for _ in range(64)
         )
         self.base32_string_seed = random_base32_string
-        self.qr_code_png = os.urandom(64)  # this would be a generated PNG
+        self.qr_code_png = random.randbytes(64)
 
         self.enable_date: Optional[datetime] = None
         self.user_attribute: Optional[dict[str, Any]] = None

moto/kms/models.py

@@ -1,5 +1,4 @@
 import json
-import os
 from collections.abc import Iterable
 from copy import copy
 from datetime import datetime, timedelta
@@ -629,7 +628,6 @@ def re_encrypt(
 
     def generate_random(self) -> None:
         # Marker to indicate this is implemented
-        # Responses uses 'os.urandom'
         pass
 
     def generate_data_key(
@@ -650,7 +648,7 @@ def generate_data_key(
         else:
             plaintext_len = number_of_bytes
 
-        plaintext = os.urandom(plaintext_len)
+        plaintext = mock_random.randbytes(plaintext_len)
 
         ciphertext_blob, arn = self.encrypt(
             key_id=key_id, plaintext=plaintext, encryption_context=encryption_context

moto/kms/responses.py

@@ -1,11 +1,11 @@
 import base64
 import json
-import os
 import re
 from typing import Any
 
 from moto.core.responses import BaseResponse
 from moto.kms.utils import RESERVED_ALIASE_TARGET_KEY_IDS, RESERVED_ALIASES
+from moto.moto_api._internal import mock_random as random
 from moto.utilities.utils import get_partition
 
 from .exceptions import (
@@ -632,7 +632,7 @@ def generate_random(self) -> str:
                 "equal to 1024"
             )
 
-        entropy = os.urandom(number_of_bytes)
+        entropy = random.randbytes(number_of_bytes)
 
         response_entropy = base64.b64encode(entropy).decode("utf-8")
 

moto/kms/utils.py

@@ -2,7 +2,6 @@
 import hashlib
 import hmac
 import io
-import os
 import struct
 from abc import ABCMeta, abstractmethod
 from collections import namedtuple
@@ -133,7 +132,7 @@ def generate_key_id(multi_region: bool = False) -> str:
 
 def generate_data_key(number_of_bytes: int) -> bytes:
     """Generate a data key."""
-    return os.urandom(number_of_bytes)
+    return mock_random.randbytes(number_of_bytes)
 
 
 def generate_master_key() -> bytes:
@@ -381,7 +380,7 @@ def encrypt(
             "1 validation error detected: Value at 'plaintext' failed to satisfy constraint: Member must have length less than or equal to 4096"
         )
 
-    iv = os.urandom(IV_LEN)
+    iv = mock_random.randbytes(IV_LEN)
     aad = _serialize_encryption_context(encryption_context=encryption_context)
 
     encryptor = Cipher(

moto/s3/models.py

@@ -6,7 +6,6 @@
 import gzip
 import itertools
 import json
-import os
 import string
 import sys
 import tempfile
@@ -454,7 +453,7 @@ def __init__(
         self.acl = acl
         self.parts: dict[int, FakeKey] = {}
         self.partlist: list[int] = []  # ordered list of part ID's
-        rand_b64 = base64.b64encode(os.urandom(UPLOAD_ID_BYTES))
+        rand_b64 = base64.b64encode(random.randbytes(UPLOAD_ID_BYTES))
         self.id = (
             rand_b64.decode("utf-8").replace("=", "").replace("+", "").replace("/", "")
         )

moto/sts/utils.py

@@ -1,5 +1,4 @@
 import base64
-import os
 import string
 
 from moto.moto_api._internal import mock_random as random
@@ -13,7 +12,7 @@
 def random_session_token() -> str:
     return (
         SESSION_TOKEN_PREFIX
-        + base64.b64encode(os.urandom(266))[len(SESSION_TOKEN_PREFIX) :].decode()
+        + base64.b64encode(random.randbytes(266))[len(SESSION_TOKEN_PREFIX) :].decode()
     )
 
 

tests/test_moto_api/seeder/test_seeder.py

@@ -41,7 +41,7 @@ def test_0(self) -> None:
         instances = self.ec2_client.run_instances(MaxCount=1, MinCount=1)["Instances"]
 
         instance_ids = [instance["InstanceId"] for instance in instances]
-        assert instance_ids == ["i-6e8b5359309cc6273"]
+        assert instance_ids == ["i-dbe4d58fed8a728e7"]
 
     def test_1(self) -> None:
         # Create some data in a different account (111111111111)

tests/test_s3/test_s3_content_encoding.py

@@ -1,9 +1,10 @@
-import os
 from tempfile import TemporaryFile
 
 import boto3
 import pytest
 
+from moto.moto_api._internal import mock_random as random
+
 from . import s3_aws_verified
 
 
@@ -22,7 +23,7 @@ def test_content_encoding_is_returned(encoding, bucket_name=None):
         attrs["ContentEncoding"] = encoding
 
     with TemporaryFile() as f:
-        f.write(os.urandom(1024))
+        f.write(random.randbytes(1024))
         f.flush()
         client.upload_fileobj(f, bucket_name, s3_file, ExtraArgs=attrs)
 

tests/test_s3/test_s3_multipart.py

@@ -12,6 +12,7 @@
 
 import moto.s3.models as s3model
 from moto import mock_aws, settings
+from moto.moto_api._internal import mock_random as random
 from moto.s3.responses import DEFAULT_REGION_NAME
 from moto.settings import (
     S3_UPLOAD_PART_MIN_SIZE,
@@ -55,14 +56,14 @@ def test_default_key_buffer_size():
     os.environ["MOTO_S3_DEFAULT_KEY_BUFFER_SIZE"] = "2"  # 2 bytes
     assert get_s3_default_key_buffer_size() == 2
     fake_key = s3model.FakeKey(
-        "a", os.urandom(1), account_id=DEFAULT_ACCOUNT_ID, region_name="us-east-1"
+        "a", random.randbytes(1), account_id=DEFAULT_ACCOUNT_ID, region_name="us-east-1"
     )
     assert fake_key._value_buffer._rolled is False
 
     os.environ["MOTO_S3_DEFAULT_KEY_BUFFER_SIZE"] = "1"  # 1 byte
     assert get_s3_default_key_buffer_size() == 1
     fake_key = s3model.FakeKey(
-        "a", os.urandom(3), account_id=DEFAULT_ACCOUNT_ID, region_name="us-east-1"
+        "a", random.randbytes(3), account_id=DEFAULT_ACCOUNT_ID, region_name="us-east-1"
     )
     assert fake_key._value_buffer._rolled is True
 
@@ -545,7 +546,7 @@ def test_multipart_upload_with_tags():
     upload = boto3.resource("s3").MultipartUpload(bucket, key, response["UploadId"])
     parts = [
         {
-            "ETag": upload.Part(i).upload(Body=os.urandom(5 * (2**20)))["ETag"],
+            "ETag": upload.Part(i).upload(Body=random.randbytes(5 * (2**20)))["ETag"],
             "PartNumber": i,
         }
         for i in range(1, 3)