healthchecks/hc/accounts/views.py

import uuid
import re

from django.conf import settings
from django.contrib import messages
from django.contrib.auth import login as auth_login
from django.contrib.auth import logout as auth_logout
from django.contrib.auth import authenticate
from django.contrib.auth.decorators import login_required
from django.contrib.auth.hashers import check_password
from django.contrib.auth.models import User
from django.core import signing
from django.http import HttpResponseForbidden, HttpResponseBadRequest
from django.shortcuts import redirect, render
from django.views.decorators.http import require_POST
from hc.accounts.forms import (EmailPasswordForm, InviteTeamMemberForm,
                               RemoveTeamMemberForm, ReportSettingsForm,
                               SetPasswordForm, TeamNameForm)
from hc.accounts.models import Profile, Member
from hc.api.models import Channel, Check
from hc.lib.badges import get_badge_url
from hc.payments.models import Subscription


def _make_user(email):
    username = str(uuid.uuid4())[:30]
    user = User(username=username, email=email)
    user.set_unusable_password()
    user.save()

    # Ensure a profile gets created
    Profile.objects.for_user(user)

    channel = Channel()
    channel.user = user
    channel.kind = "email"
    channel.value = email
    channel.email_verified = True
    channel.save()

    return user


def _associate_demo_check(request, user):
    if "welcome_code" not in request.session:
        return

    try:
        check = Check.objects.get(code=request.session["welcome_code"])
    except Check.DoesNotExist:
        return

    # Only associate demo check if it doesn't have an owner already.
    if check.user:
        return

    check.user = user
    check.save()

    check.assign_all_channels()

    del request.session["welcome_code"]


def login(request, show_password=False):
    bad_credentials = False
    if request.method == 'POST':
        form = EmailPasswordForm(request.POST)
        if form.is_valid():
            email = form.cleaned_data["email"]
            password = form.cleaned_data["password"]
            if len(password):
                user = authenticate(username=email, password=password)
                if user is not None and user.is_active:
                    auth_login(request, user)
                    return redirect("hc-checks")
                bad_credentials = True
                show_password = True
            else:
                user = None
                try:
                    user = User.objects.get(email=email)
                except User.DoesNotExist:
                    if settings.REGISTRATION_OPEN:
                        user = _make_user(email)
                        _associate_demo_check(request, user)
                    else:
                        bad_credentials = True

                if user:
                    profile = Profile.objects.for_user(user)
                    profile.send_instant_login_link()
                    return redirect("hc-login-link-sent")

    else:
        form = EmailPasswordForm()

    bad_link = request.session.pop("bad_link", None)
    ctx = {
        "form": form,
        "bad_credentials": bad_credentials,
        "bad_link": bad_link,
        "show_password": show_password
    }
    return render(request, "accounts/login.html", ctx)


def logout(request):
    auth_logout(request)
    return redirect("hc-index")


def login_link_sent(request):
    return render(request, "accounts/login_link_sent.html")


def set_password_link_sent(request):
    return render(request, "accounts/set_password_link_sent.html")


def check_token(request, username, token):
    if request.user.is_authenticated and request.user.username == username:
        # User is already logged in
        return redirect("hc-checks")

    # Some email servers open links in emails to check for malicious content.
    # To work around this, we sign user in if the method is POST.
    #
    # If the method is GET, we instead serve a HTML form and a piece
    # of Javascript to automatically submit it.

    if request.method == "POST":
        user = authenticate(username=username, token=token)
        if user is not None and user.is_active:
            # This should get rid of "welcome_code" in session
            request.session.flush()

            user.profile.token = ""
            user.profile.save()
            auth_login(request, user)

            return redirect("hc-checks")

        request.session["bad_link"] = True
        return redirect("hc-login")

    return render(request, "accounts/check_token_submit.html")


@login_required
def profile(request):
    profile = request.user.profile
    # Switch user back to its own team
    if request.team != profile:
        request.team = profile
        profile.current_team = profile
        profile.save()

    show_api_key = False
    if request.method == "POST":
        if "set_password" in request.POST:
            profile.send_set_password_link()
            return redirect("hc-set-password-link-sent")
        elif "create_api_key" in request.POST:
            profile.set_api_key()
            show_api_key = True
            messages.success(request, "The API key has been created!")
        elif "revoke_api_key" in request.POST:
            profile.api_key = ""
            profile.save()
            messages.info(request, "The API key has been revoked!")
        elif "show_api_key" in request.POST:
            show_api_key = True
        elif "invite_team_member" in request.POST:
            if not profile.team_access_allowed:
                return HttpResponseForbidden()

            form = InviteTeamMemberForm(request.POST)
            if form.is_valid():

                email = form.cleaned_data["email"]
                try:
                    user = User.objects.get(email=email)
                except User.DoesNotExist:
                    user = _make_user(email)

                profile.invite(user)
                messages.success(request, "Invitation to %s sent!" % email)
        elif "remove_team_member" in request.POST:
            form = RemoveTeamMemberForm(request.POST)
            if form.is_valid():

                email = form.cleaned_data["email"]
                farewell_user = User.objects.get(email=email)
                farewell_user.profile.current_team = None
                farewell_user.profile.save()

                Member.objects.filter(team=profile,
                                      user=farewell_user).delete()

                messages.info(request, "%s removed from team!" % email)
        elif "set_team_name" in request.POST:
            if not profile.team_access_allowed:
                return HttpResponseForbidden()

            form = TeamNameForm(request.POST)
            if form.is_valid():
                profile.team_name = form.cleaned_data["team_name"]
                profile.save()
                messages.success(request, "Team Name updated!")

    ctx = {
        "page": "profile",
        "profile": profile,
        "show_api_key": show_api_key
    }

    return render(request, "accounts/profile.html", ctx)


@login_required
def notifications(request):
    profile = request.user.profile
    # Switch user back to its default team
    if profile.current_team_id != profile.id:
        request.team = profile
        profile.current_team_id = profile.id
        profile.save()

    if request.method == "POST":
        form = ReportSettingsForm(request.POST)
        if form.is_valid():
            profile.reports_allowed = form.cleaned_data["reports_allowed"]
            profile.save()
            messages.success(request, "Your settings have been updated!")

    ctx = {
        "page": "profile",
        "profile": profile,
    }

    return render(request, "accounts/notifications.html", ctx)


@login_required
def badges(request):
    profile = request.user.profile
    # Switch user back to its own team
    if request.team != profile:
        request.team = profile
        profile.current_team = profile
        profile.save()

    tags = set()
    for check in Check.objects.filter(user=request.team.user):
        tags.update(check.tags_list())

    username = request.team.user.username
    urls = []
    for tag in sorted(tags, key=lambda s: s.lower()):
        if not re.match("^[\w-]+$", tag):
            continue

        urls.append({
            "svg": get_badge_url(username, tag),
            "json": get_badge_url(username, tag, format="json"),
        })

    ctx = {
        "page": "profile",
        "urls": urls
    }

    return render(request, "accounts/badges.html", ctx)


@login_required
def set_password(request, token):
    profile = request.user.profile
    if not check_password(token, profile.token):
        return HttpResponseBadRequest()

    if request.method == "POST":
        form = SetPasswordForm(request.POST)
        if form.is_valid():
            password = form.cleaned_data["password"]
            request.user.set_password(password)
            request.user.save()

            profile.token = ""
            profile.save()

            # Setting a password logs the user out, so here we
            # log them back in.
            u = authenticate(username=request.user.email, password=password)
            auth_login(request, u)

            messages.success(request, "Your password has been set!")
            return redirect("hc-profile")

    return render(request, "accounts/set_password.html", {})


def unsubscribe_reports(request, username):
    try:
        signing.Signer().unsign(request.GET.get("token"))
    except signing.BadSignature:
        return HttpResponseBadRequest()

    user = User.objects.get(username=username)
    user.profile.reports_allowed = False
    user.profile.save()

    return render(request, "accounts/unsubscribed.html")


@login_required
def switch_team(request, target_username):
    try:
        other_user = User.objects.get(username=target_username)
    except User.DoesNotExist:
        return HttpResponseForbidden()

    # The rules:
    # Superuser can switch to any team.
    access_ok = request.user.is_superuser

    # Users can switch to their own teams.
    if not access_ok and other_user.id == request.user.id:
        access_ok = True

    # Users can switch to teams they are members of.
    if not access_ok:
        for membership in request.user.member_set.all():
            if membership.team.user.id == other_user.id:
                access_ok = True
                break

    if not access_ok:
        return HttpResponseForbidden()

    request.user.profile.current_team = other_user.profile
    request.user.profile.save()

    return redirect("hc-checks")


@require_POST
@login_required
def close(request):
    user = request.user

    # Subscription needs to be canceled before it is deleted:
    sub = Subscription.objects.filter(user=user).first()
    if sub:
        sub.cancel()

    user.delete()

    # Deleting user also deletes its profile, checks, channels etc.

    request.session.flush()
    return redirect("hc-index")