From 5f47d172e0ab101cd107b4bbe7d1f0b314825de0 Mon Sep 17 00:00:00 2001 From: Son Date: Sun, 28 Nov 2021 11:51:31 +0100 Subject: [PATCH] move fake_data() to dedicated fiel --- app/fake_data.py | 271 +++++++++++++++++++++++++++++++++++++++++++++++ server.py | 251 +------------------------------------------ 2 files changed, 273 insertions(+), 249 deletions(-) create mode 100644 app/fake_data.py diff --git a/app/fake_data.py b/app/fake_data.py new file mode 100644 index 00000000..b5b0d0b9 --- /dev/null +++ b/app/fake_data.py @@ -0,0 +1,271 @@ +import os + +import arrow + +from app import s3 +from app.config import ROOT_DIR, get_abs_path, FIRST_ALIAS_DOMAIN +from app.db import Session +from app.log import LOG +from app.models import ( + User, + File, + Alias, + RefusedEmail, + Contact, + EmailLog, + LifetimeCoupon, + Coupon, + Subscription, + PlanEnum, + CoinbaseSubscription, + ApiKey, + Mailbox, + AliasMailbox, + CustomDomain, + Directory, + Client, + RedirectUri, + ClientUser, + Referral, + Payout, + Notification, + ManualSubscription, + SLDomain, + Hibp, + AliasHibp, +) +from app.pgp_utils import load_public_key + + +def fake_data(): + LOG.d("create fake data") + + # Create a user + user = User.create( + email="john@wick.com", + name="John Wick", + password="password", + activated=True, + is_admin=True, + # enable_otp=True, + otp_secret="base32secret3232", + intro_shown=True, + fido_uuid=None, + ) + user.trial_end = None + Session.commit() + + # add a profile picture + file_path = "profile_pic.svg" + s3.upload_from_bytesio( + file_path, + open(os.path.join(ROOT_DIR, "static", "default-icon.svg"), "rb"), + content_type="image/svg", + ) + file = File.create(user_id=user.id, path=file_path, commit=True) + user.profile_picture_id = file.id + Session.commit() + + # create a bounced email + alias = Alias.create_new_random(user) + Session.commit() + + bounce_email_file_path = "bounce.eml" + s3.upload_email_from_bytesio( + bounce_email_file_path, + open(os.path.join(ROOT_DIR, "local_data", "email_tests", "2.eml"), "rb"), + "download.eml", + ) + refused_email = RefusedEmail.create( + path=bounce_email_file_path, + full_report_path=bounce_email_file_path, + user_id=user.id, + commit=True, + ) + + contact = Contact.create( + user_id=user.id, + alias_id=alias.id, + website_email="hey@google.com", + reply_email="rep@sl.local", + commit=True, + ) + EmailLog.create( + user_id=user.id, + contact_id=contact.id, + alias_id=contact.alias_id, + refused_email_id=refused_email.id, + bounced=True, + commit=True, + ) + + LifetimeCoupon.create(code="lifetime-coupon", nb_used=10, commit=True) + Coupon.create(code="coupon", commit=True) + + # Create a subscription for user + Subscription.create( + user_id=user.id, + cancel_url="https://checkout.paddle.com/subscription/cancel?user=1234", + update_url="https://checkout.paddle.com/subscription/update?user=1234", + subscription_id="123", + event_time=arrow.now(), + next_bill_date=arrow.now().shift(days=10).date(), + plan=PlanEnum.monthly, + commit=True, + ) + + CoinbaseSubscription.create( + user_id=user.id, end_at=arrow.now().shift(days=10), commit=True + ) + + api_key = ApiKey.create(user_id=user.id, name="Chrome") + api_key.code = "code" + + api_key = ApiKey.create(user_id=user.id, name="Firefox") + api_key.code = "codeFF" + + pgp_public_key = open(get_abs_path("local_data/public-pgp.asc")).read() + m1 = Mailbox.create( + user_id=user.id, + email="pgp@example.org", + verified=True, + pgp_public_key=pgp_public_key, + ) + m1.pgp_finger_print = load_public_key(pgp_public_key) + Session.commit() + + # example@example.com is in a LOT of data breaches + Alias.create(email="example@example.com", user_id=user.id, mailbox_id=m1.id) + + for i in range(3): + if i % 2 == 0: + a = Alias.create( + email=f"e{i}@{FIRST_ALIAS_DOMAIN}", user_id=user.id, mailbox_id=m1.id + ) + else: + a = Alias.create( + email=f"e{i}@{FIRST_ALIAS_DOMAIN}", + user_id=user.id, + mailbox_id=user.default_mailbox_id, + ) + Session.commit() + + if i % 5 == 0: + if i % 2 == 0: + AliasMailbox.create(alias_id=a.id, mailbox_id=user.default_mailbox_id) + else: + AliasMailbox.create(alias_id=a.id, mailbox_id=m1.id) + Session.commit() + + # some aliases don't have any activity + # if i % 3 != 0: + # contact = Contact.create( + # user_id=user.id, + # alias_id=a.id, + # website_email=f"contact{i}@example.com", + # reply_email=f"rep{i}@sl.local", + # ) + # Session.commit() + # for _ in range(3): + # EmailLog.create(user_id=user.id, contact_id=contact.id, alias_id=contact.alias_id) + # Session.commit() + + # have some disabled alias + if i % 5 == 0: + a.enabled = False + Session.commit() + + custom_domain1 = CustomDomain.create(user_id=user.id, domain="ab.cd", verified=True) + Session.commit() + + Alias.create( + user_id=user.id, + email="first@ab.cd", + mailbox_id=user.default_mailbox_id, + custom_domain_id=custom_domain1.id, + commit=True, + ) + + Alias.create( + user_id=user.id, + email="second@ab.cd", + mailbox_id=user.default_mailbox_id, + custom_domain_id=custom_domain1.id, + commit=True, + ) + + Directory.create(user_id=user.id, name="abcd") + Directory.create(user_id=user.id, name="xyzt") + Session.commit() + + # Create a client + client1 = Client.create_new(name="Demo", user_id=user.id) + client1.oauth_client_id = "client-id" + client1.oauth_client_secret = "client-secret" + Session.commit() + + RedirectUri.create( + client_id=client1.id, uri="https://your-website.com/oauth-callback" + ) + + client2 = Client.create_new(name="Demo 2", user_id=user.id) + client2.oauth_client_id = "client-id2" + client2.oauth_client_secret = "client-secret2" + Session.commit() + + ClientUser.create(user_id=user.id, client_id=client1.id, name="Fake Name") + + referral = Referral.create(user_id=user.id, code="Website", name="First referral") + Referral.create(user_id=user.id, code="Podcast", name="First referral") + Payout.create( + user_id=user.id, amount=1000, number_upgraded_account=100, payment_method="BTC" + ) + Payout.create( + user_id=user.id, + amount=5000, + number_upgraded_account=200, + payment_method="PayPal", + ) + Session.commit() + + for i in range(6): + Notification.create(user_id=user.id, message=f"""Hey hey {i} """ * 10) + Session.commit() + + user2 = User.create( + email="winston@continental.com", + password="password", + activated=True, + referral_id=referral.id, + ) + Mailbox.create(user_id=user2.id, email="winston2@high.table", verified=True) + Session.commit() + + ManualSubscription.create( + user_id=user2.id, + end_at=arrow.now().shift(years=1, days=1), + comment="Local manual", + commit=True, + ) + + SLDomain.create(domain="premium.com", premium_only=True, commit=True) + + hibp1 = Hibp.create( + name="first breach", description="breach description", commit=True + ) + hibp2 = Hibp.create( + name="second breach", description="breach description", commit=True + ) + breached_alias1 = Alias.create( + email="john@example.com", user_id=user.id, mailbox_id=m1.id, commit=True + ) + breached_alias2 = Alias.create( + email="wick@example.com", user_id=user.id, mailbox_id=m1.id, commit=True + ) + AliasHibp.create(hibp_id=hibp1.id, alias_id=breached_alias1.id) + AliasHibp.create(hibp_id=hibp2.id, alias_id=breached_alias2.id) + + # old domain will have ownership_verified=True + CustomDomain.create( + user_id=user.id, domain="old.com", verified=True, ownership_verified=True + ) diff --git a/server.py b/server.py index 198d724b..9e6e110a 100644 --- a/server.py +++ b/server.py @@ -26,7 +26,7 @@ from sentry_sdk.integrations.flask import FlaskIntegration from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration from werkzeug.middleware.proxy_fix import ProxyFix -from app import paddle_utils, s3, config +from app import paddle_utils, config from app.admin_model import ( SLAdminIndexView, UserAdmin, @@ -63,12 +63,10 @@ from app.config import ( LANDING_PAGE_URL, STATUS_PAGE_URL, SUPPORT_EMAIL, - get_abs_path, PADDLE_MONTHLY_PRODUCT_IDS, PADDLE_YEARLY_PRODUCT_IDS, PGP_SIGNER, COINBASE_WEBHOOK_SECRET, - ROOT_DIR, PAGE_LIMIT, PADDLE_COUPON_ID, ) @@ -78,39 +76,28 @@ from app.developer.base import developer_bp from app.discover.base import discover_bp from app.email_utils import send_email, render from app.extensions import login_manager, migrate, limiter +from app.fake_data import fake_data from app.jose_utils import get_jwk_key from app.log import LOG from app.models import ( Client, User, - ClientUser, Alias, - RedirectUri, Subscription, PlanEnum, - ApiKey, CustomDomain, LifetimeCoupon, - Directory, Mailbox, Referral, - AliasMailbox, - Notification, CoinbaseSubscription, EmailLog, - File, Contact, - RefusedEmail, ManualSubscription, Payout, Coupon, - SLDomain, - Hibp, - AliasHibp, ) from app.monitor.base import monitor_bp from app.oauth.base import oauth_bp -from app.pgp_utils import load_public_key from app.utils import random_string if SENTRY_DSN: @@ -213,240 +200,6 @@ def create_app() -> Flask: return app -def fake_data(): - LOG.d("create fake data") - - # Create a user - user = User.create( - email="john@wick.com", - name="John Wick", - password="password", - activated=True, - is_admin=True, - # enable_otp=True, - otp_secret="base32secret3232", - intro_shown=True, - fido_uuid=None, - ) - user.trial_end = None - Session.commit() - - # add a profile picture - file_path = "profile_pic.svg" - s3.upload_from_bytesio( - file_path, - open(os.path.join(ROOT_DIR, "static", "default-icon.svg"), "rb"), - content_type="image/svg", - ) - file = File.create(user_id=user.id, path=file_path, commit=True) - user.profile_picture_id = file.id - Session.commit() - - # create a bounced email - alias = Alias.create_new_random(user) - Session.commit() - - bounce_email_file_path = "bounce.eml" - s3.upload_email_from_bytesio( - bounce_email_file_path, - open(os.path.join(ROOT_DIR, "local_data", "email_tests", "2.eml"), "rb"), - "download.eml", - ) - refused_email = RefusedEmail.create( - path=bounce_email_file_path, - full_report_path=bounce_email_file_path, - user_id=user.id, - commit=True, - ) - - contact = Contact.create( - user_id=user.id, - alias_id=alias.id, - website_email="hey@google.com", - reply_email="rep@sl.local", - commit=True, - ) - EmailLog.create( - user_id=user.id, - contact_id=contact.id, - alias_id=contact.alias_id, - refused_email_id=refused_email.id, - bounced=True, - commit=True, - ) - - LifetimeCoupon.create(code="lifetime-coupon", nb_used=10, commit=True) - Coupon.create(code="coupon", commit=True) - - # Create a subscription for user - Subscription.create( - user_id=user.id, - cancel_url="https://checkout.paddle.com/subscription/cancel?user=1234", - update_url="https://checkout.paddle.com/subscription/update?user=1234", - subscription_id="123", - event_time=arrow.now(), - next_bill_date=arrow.now().shift(days=10).date(), - plan=PlanEnum.monthly, - commit=True, - ) - - CoinbaseSubscription.create( - user_id=user.id, end_at=arrow.now().shift(days=10), commit=True - ) - - api_key = ApiKey.create(user_id=user.id, name="Chrome") - api_key.code = "code" - - api_key = ApiKey.create(user_id=user.id, name="Firefox") - api_key.code = "codeFF" - - pgp_public_key = open(get_abs_path("local_data/public-pgp.asc")).read() - m1 = Mailbox.create( - user_id=user.id, - email="pgp@example.org", - verified=True, - pgp_public_key=pgp_public_key, - ) - m1.pgp_finger_print = load_public_key(pgp_public_key) - Session.commit() - - # example@example.com is in a LOT of data breaches - Alias.create(email="example@example.com", user_id=user.id, mailbox_id=m1.id) - - for i in range(3): - if i % 2 == 0: - a = Alias.create( - email=f"e{i}@{FIRST_ALIAS_DOMAIN}", user_id=user.id, mailbox_id=m1.id - ) - else: - a = Alias.create( - email=f"e{i}@{FIRST_ALIAS_DOMAIN}", - user_id=user.id, - mailbox_id=user.default_mailbox_id, - ) - Session.commit() - - if i % 5 == 0: - if i % 2 == 0: - AliasMailbox.create(alias_id=a.id, mailbox_id=user.default_mailbox_id) - else: - AliasMailbox.create(alias_id=a.id, mailbox_id=m1.id) - Session.commit() - - # some aliases don't have any activity - # if i % 3 != 0: - # contact = Contact.create( - # user_id=user.id, - # alias_id=a.id, - # website_email=f"contact{i}@example.com", - # reply_email=f"rep{i}@sl.local", - # ) - # Session.commit() - # for _ in range(3): - # EmailLog.create(user_id=user.id, contact_id=contact.id, alias_id=contact.alias_id) - # Session.commit() - - # have some disabled alias - if i % 5 == 0: - a.enabled = False - Session.commit() - - custom_domain1 = CustomDomain.create(user_id=user.id, domain="ab.cd", verified=True) - Session.commit() - - Alias.create( - user_id=user.id, - email="first@ab.cd", - mailbox_id=user.default_mailbox_id, - custom_domain_id=custom_domain1.id, - commit=True, - ) - - Alias.create( - user_id=user.id, - email="second@ab.cd", - mailbox_id=user.default_mailbox_id, - custom_domain_id=custom_domain1.id, - commit=True, - ) - - Directory.create(user_id=user.id, name="abcd") - Directory.create(user_id=user.id, name="xyzt") - Session.commit() - - # Create a client - client1 = Client.create_new(name="Demo", user_id=user.id) - client1.oauth_client_id = "client-id" - client1.oauth_client_secret = "client-secret" - Session.commit() - - RedirectUri.create( - client_id=client1.id, uri="https://your-website.com/oauth-callback" - ) - - client2 = Client.create_new(name="Demo 2", user_id=user.id) - client2.oauth_client_id = "client-id2" - client2.oauth_client_secret = "client-secret2" - Session.commit() - - ClientUser.create(user_id=user.id, client_id=client1.id, name="Fake Name") - - referral = Referral.create(user_id=user.id, code="Website", name="First referral") - Referral.create(user_id=user.id, code="Podcast", name="First referral") - Payout.create( - user_id=user.id, amount=1000, number_upgraded_account=100, payment_method="BTC" - ) - Payout.create( - user_id=user.id, - amount=5000, - number_upgraded_account=200, - payment_method="PayPal", - ) - Session.commit() - - for i in range(6): - Notification.create(user_id=user.id, message=f"""Hey hey {i} """ * 10) - Session.commit() - - user2 = User.create( - email="winston@continental.com", - password="password", - activated=True, - referral_id=referral.id, - ) - Mailbox.create(user_id=user2.id, email="winston2@high.table", verified=True) - Session.commit() - - ManualSubscription.create( - user_id=user2.id, - end_at=arrow.now().shift(years=1, days=1), - comment="Local manual", - commit=True, - ) - - SLDomain.create(domain="premium.com", premium_only=True, commit=True) - - hibp1 = Hibp.create( - name="first breach", description="breach description", commit=True - ) - hibp2 = Hibp.create( - name="second breach", description="breach description", commit=True - ) - breached_alias1 = Alias.create( - email="john@example.com", user_id=user.id, mailbox_id=m1.id, commit=True - ) - breached_alias2 = Alias.create( - email="wick@example.com", user_id=user.id, mailbox_id=m1.id, commit=True - ) - AliasHibp.create(hibp_id=hibp1.id, alias_id=breached_alias1.id) - AliasHibp.create(hibp_id=hibp2.id, alias_id=breached_alias2.id) - - # old domain will have ownership_verified=True - CustomDomain.create( - user_id=user.id, domain="old.com", verified=True, ownership_verified=True - ) - - @login_manager.user_loader def load_user(alternative_id): user = User.get_by(alternative_id=alternative_id)