Patched code sample

# In the context of python-social-auth, the authentication flow is managed
# by a "pipeline" of functions. This code simulates a key function within
# that pipeline, `get_user`, to demonstrate how the vulnerability was fixed.
#
# The vulnerability: The system would associate a social account with a local
# account by email, even if the explicit 'associate_by_email' step was not
# in the configured pipeline.
#
# The fix: Ensure the email association logic is only executed if
# 'associate_by_email' is explicitly defined in the settings pipeline.

# --- Mocks and Simulation Setup ---

class MockUser:
    """A mock user model."""
    def __init__(self, email, username):
        self.email = email
        self.username = username

    def __repr__(self):
        return f"<User: {self.username} ({self.email})>"

class MockUserModelManager:
    """A mock manager to find users."""
    _users = []

    def get_by_email(self, email):
        if not email:
            return None
        for user in self._users:
            if user.email.lower() == email.lower():
                return user
        return None

# A simplified representation of the application's user model
User = MockUserModelManager()

class MockStrategy:
    """A mock strategy object that holds settings."""
    def __init__(self, pipeline):
        self.settings = {'SOCIAL_AUTH_PIPELINE': pipeline}

    def setting(self, name):
        return self.settings.get(name)

    def get_user_model(self):
        return User

# --- Code Representing the Vulnerability ---

def get_user_vulnerable(strategy, details, user=None, **kwargs):
    """
    This function simulates the vulnerable behavior.
    It attempts to find a user by email regardless of the pipeline configuration.
    """
    if user:
        return {'user': user}

    # VULNERABLE BEHAVIOR:
    # This block unconditionally tries to find a user by email. An attacker
    # could create a social account with a victim's unverified email,
    # log in, and this code would incorrectly associate them with the
    # victim's existing account.
    user_model = strategy.get_user_model()
    email = details.get('email')
    if email:
        user = user_model.get_by_email(email)
        if user:
            return {'user': user}

    return {'user': None}


# --- Code Representing the Fix ---

def get_user_fixed(strategy, details, user=None, **kwargs):
    """
    This function simulates the patched behavior.
    It only attempts to find a user by email if the corresponding step
    is present in the pipeline configuration.
    """
    if user:
        return {'user': user}

    # THE FIX:
    # The code now checks if the email association logic is explicitly
    # enabled in the configured pipeline before executing it.
    pipeline = strategy.setting('SOCIAL_AUTH_PIPELINE') or []
    
    # This simulates the name of the function/step responsible for email association.
    associate_by_email_step = 'social_core.pipeline.user.associate_by_email'

    if associate_by_email_step in pipeline:
        user_model = strategy.get_user_model()
        email = details.get('email')
        if email:
            user = user_model.get_by_email(email)
            if user:
                return {'user': user}

    return {'user': None}


# --- Demonstration ---

if __name__ == '__main__':
    # 1. Setup a pre-existing user in the database.
    existing_user = MockUser(email='victim@example.com', username='victim')
    User._users.append(existing_user)
    print(f"[*] A user exists in the database: {existing_user}\n")

    # 2. An attacker provides details from a social provider. The provider did
    #    not verify the email, so the attacker could claim 'victim@example.com'.
    attacker_social_details = {
        'username': 'attacker_on_social_site',
        'email': 'victim@example.com'
    }
    print(f"[*] Attacker logs in via a social provider with details: {attacker_social_details}\n")

    # 3. Define a pipeline that *does not* include the email association step.
    #    This configuration implies the site owner does not trust incoming emails.
    secure_pipeline_config = (
        'social_core.pipeline.social_auth.social_details',
        'social_core.pipeline.social_auth.social_uid',
        'social_core.pipeline.social_auth.auth_allowed',
        'social_core.pipeline.social_auth.social_user',
        'social_core.pipeline.user.create_user',
        'social_core.pipeline.social_auth.associate_user',
        'social_core.pipeline.social_auth.load_extra_data',
        'social_core.pipeline.user.user_details',
    )
    
    # 4. Create a strategy instance with this secure pipeline.
    strategy = MockStrategy(pipeline=secure_pipeline_config)
    print(f"[*] The application is configured with a pipeline that *excludes* email association.\n")

    # 5. Run the VULNERABLE version.
    print("--- Running VULNERABLE Code ---")
    vulnerable_result = get_user_vulnerable(strategy, attacker_social_details)
    found_user_vulnerable = vulnerable_result.get('user')

    if found_user_vulnerable:
        print(f"VULNERABLE: SUCCESS. Attacker was associated with user '{found_user_vulnerable.username}'. ACCOUNT COMPROMISED.")
    else:
        print("VULNERABLE: FAILED. No user was found (this is unexpected for the vulnerable code).")
    
    print("-" * 30 + "\n")

    # 6. Run the FIXED version with the exact same setup.
    print("--- Running FIXED Code ---")
    fixed_result = get_user_fixed(strategy, attacker_social_details)
    found_user_fixed = fixed_result.get('user')

    if found_user_fixed:
        print(f"FIXED: FAILED. Attacker was incorrectly associated with user '{found_user_fixed.username}'.")
    else:
        print(f"FIXED: SUCCESS. No existing user was found by email. A new account would be created for the attacker instead. ACCOUNT SECURE.")