ShahriyarR

from flask import Flask, request
import threading
import time

app = Flask(__name__)

# Mock database storing user data
users = {
    '1': {'email': 'user1@example.com'},
    '2': {'email': 'user2@example.com'}

ShahriyarR

import requests
import threading

def send_request(user_id, new_email):
    response = requests.post(f'http://127.0.0.1:5000/change_email/{user_id}', data={'email': new_email})
    print(response.text)

# Create threads to simulate concurrent requests
thread1 = threading.Thread(target=send_request, args=('1', 'newemail1@example.com'))
thread2 = threading.Thread(target=send_request, args=('2', 'newemail2@example.com'))

ShahriyarR

@icontract.invariant(
    lambda self: len(self) == 0 or len(self) == 1,
    "There can be no or only single secret data stored",
)
class ReadOnce(metaclass=Final):
    ...

ShahriyarR

@icontract.ensure(lambda self: not self.__secrets and not self.__is_consumed)
def __init__(self) -> None:
    self.__reset_secrets()
    self.__reset_is_consumed()

ShahriyarR

import re

def validate_password(password: str) -> bool:
    reg = r"^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[@$!%*#?&])[A-Za-z\d@$!#%*?&]{6,20}$"
    pattern = re.compile(reg)
    return bool(re.search(pattern, password))

ShahriyarR

import icontract
from readonce import ReadOnce

def validate_password_length(password: str) -> bool:
    return len(password) > 7

class Password(ReadOnce):
    @icontract.ensure(lambda self: len(self) == 1, "Secret is missing")
    @icontract.require(
        lambda password: validate_password_length(password),

ShahriyarR

from pydantic import BaseModel, validator

class InvalidDBCredentialsModel(BaseModel):
    comment: str
    password: Password
    uri: DBUri
    port: DBPort
    host: DBHost

    @validator("password")

ShahriyarR

>>> credentials.json()
...
TypeError: Object of type 'Password' is not JSON serializable

ShahriyarR

credentials.dict()
{'comment': 'The Hacked Database', 'password': ReadOnce[secrets=*****], 'uri': ReadOnce[secrets=*****], 'port': ReadOnce[secrets=*****], 'host': ReadOnce[secrets=*****]}

ShahriyarR

>>> credentials = DBCredentialsModel(comment="The Hacked Database", password=Password("db-password"), uri=DBUri("mysql://"), port=DBPort(3306), host=DBHost("localhost"))
>>> credentials
DBCredentialsModel(comment='The Hacked Database', password=ReadOnce[secrets=*****], uri=ReadOnce[secrets=*****], port=ReadOnce[secrets=*****], host=ReadOnce[secrets=*****])