Basic Querying¶

So let’s get started. To perform a query, we need the session on which all queries are executed. Note that the DBSession is scoped to the current thread. In our context that means that calling it will always give the exact same session back as long as we are in the same thread. So it is not required to pass it around all the time. Here’s how you get a session to start querying:

from fluxscoreboard.models import DBSession
dbsession = DBSession()

Now we can execute a query, for example, get all challenges:

from fluxscoreboard.models.challenges import Challenge
all_challenges = dbsession.query(Challenge)

Note that we did not actually execute a query yet but instead have a query object. It produces results once used as an iterable, e.g. list(all_challenges) or all_challenges.all(). Then you have a list of challenges. However, here is one thing to note: You should always return query, i.e. not call all or similar to produce actual results. That is done those items are actually used. This is advantageous in two ways: For one, it supports the lazy technique by which a query is only emitted once the results are actually need. And for the other part, you can refine such queries when they are not yet results. So you could modify the query to only get challenges which are manual:

manual_challenges = all_challenges.filter(Challenge.manual == True)

Now this is a query that can be executed and it will limit the results. And you should remember to always pass around queries, never the actual lists, until they are used.

page = Page(manual_challenges, page=1,
            item_count=manual_challenges.count(), url=page_url)

Upgrading the Database¶

During development it will likely happen that you need to upgrade the database schema at some point which can lead to problems for multiple reasons. For one, we are using an ORM and since we are not reflecting, we make changes in Python code, but these are not transferred into the database automatically. Secondly, when distributing changes to other instances, the database needs to be adjusted as well. Thus, we use Alembic.

With Alembic, upgrades to the database are managed automatically. In this section, you will find a small overview of how to run default commands that cover the most basic way of doing it in our application. For anything beyond that, check out the original Alembic documentation which covers all those topics.

Suppose you made a change to your database. Before you can run alembic in any way you must do one thing first: Have a working configuration. This could be your development.ini if working locally or it could be the running production.ini. The default configuration files already contain a good working configuration. The most important part is that you already have a valid database configuration so you can actually connect to the database and make changes. Seems legit, right?

After you have done this, let’s get to work. The easiest and fastest way is to let Alembic try to detect your changes. This will work for most cases, but it will, for example, fail on renaming of columns or tables and instead detect a pair of add and delete. Anyway, this is how you let alembic create a basic file for your changes:

alembic -c development.ini revision --autogenerate -m "YOUR_CHANGE_DESCRIPTION"

Alembic will generate a file in the alembic/versions directory and give you its name. Open it and check the upgrade and downgrade methods: All you need to do is make sure that these do what you need. Afterwards, you should testrun your upgrade:

alembic -c development.ini upgrade head

This should be straightforward and you should now have an up to date database. Check that it works, i.e. run unit tests, do manual testing, etc. and then make sure downgrade works. For this, you need to find out the previous revision which is specified in the file you opened above as the variable down_revision. Lets say it is 123abc (if there’s no previous revision, specify base):

alembic -c development.ini downgrade 123abc

And that’s it. No more work has to be done here. You can now run the upgrade again to get your database up to date and work more on the application.

Defining a Basic Form¶

To define a form derive from the wtforms.form.Form class and start defining your fields (for a more thorough definition and explanation refer to the original documentation):

from wtforms.form import Form
from wtforms.fields.simple import TextField, SubmitField

class SampleForm(Form):
    text = TextField("Sample Text")

    submit = SubmitField("Submit")

Using a Form¶

Now we can use this form (and load it with data) from a view:

def some_view(request):
    form = SampleForm(request.POST)

Then we make sure it is valid:

if not form.validate():
    # do something
    pass

But what does validate do?

Defining Validation on a Form¶

To get validation, we define a list of validators on a field. For example:

from fluxscoreboard.forms.validators import required_validator
class SampleForm(Form):
    text = TextField("Sample Text",
                     validators=[required_validator]
                    )
    ...

For details on validators already provided, see the documentation for fluxscoreboard.forms.validators. There are already some common validators for length and other stuff defined. Also pay attention that you should have a validator that checks for the allowed database length (e.g. if you database column is Unicode(255) check that it is not longer than 255 characters).

From- & Database-Interaction¶

Getting a form into the database is easy.

# Create the form
form = SampleForm(request.POST)
# Create the database object
dbitem = Sample()
# Fill it with the form data
form.populate_obj(dbitem)

Of course, this only works when the form fields and database names are the same. You can also manually map the fields together if you want.

Filling a form from the database is also easy:

# Load the object (maybe from database)
dbitem = get_item()
# Create the form
form = SampleForm(request.POST, dbitem)

What does this do? It loads the database item from somewhere and then instantiates the form as previously but now it fills the missing fields from the database.

You can combine the two approaches above to retrieve a form to be edited and then save it back to the database:

def my_view(request):
    dbitem = get_item()
    form = SampleForm(request.POST, dbitem)
    if request.method == 'POST':
        if not form.validate():
            # handle it
            pass
        form.populate_obj(dbitem)
        # redirect or something
    return {'form': form}

This approach will load an item from the database, then fill the form correctly and on a GET request just display it. Once editing is done and the form gets submitted we now load the form data into the database item.

Creating a New Template¶

Simple create a file in the templates/ directory with a suffix of .mako and put the following content in it to get started:

<%inherit file="base.mako"/>

This one line integrates your template with the default style. The next step is to add the HTML you want. For the syntax of Mako look at the Mako documentation.

Page Style¶

The page is/was developed with Bootstrap 3 which provides some nice styles to work with when a design is not ready yet. You are encouraged to work with the bootstrap classes as we will work to have a stylesheet that defines the same classes and can act as a drop-in replacement thus allowing us to quickly change the style of the page when we want.

Using Defs with Forms¶

When displaying forms you will find yourself repeating over and over (either when displaying similar forms or even inside a form for each field). You can use defs to make your life a whole lot easier. For example, take this def that renders a form.

<%def name="render_form(action, form, legend, display_cancel=True)">
    <form method="POST" action="${action}" class="form-horizontal">
        <legend>${legend}</legend>
        % for field in [item for item in form if item.name not in ["id", "submit", "cancel"]]:
        <div class="form-group">
            ${field.label(class_="col-4 control-label")}
            <div class="col-8">
    ## This is a really ugly solution to a limitation of WTForms. It would be a lot nicer to rebuild the form fields so they do this automatically.
                ${field(class_="form-control")}
                % for msg in getattr(field, 'errors', []):
                    <div class="alert alert-danger">${msg}</div>
                % endfor
            </div>
        </div>
        % endfor
        <div class="col-4"></div>
        <div class="col-8">
            % if getattr(form, 'id', None) is not None:
                ${form.id()}
            % endif
            % if display_cancel and hasattr(form, 'cancel'):
                ${form.cancel(class_="btn btn-default")}
            % endif
            ${form.submit(class_="btn btn-primary")}
        </div>
    </form>
</%def>

This is a somewhat complex piece of code but it shows nicely how to build a reusable def: This will render a simple form with a legend and all fields horizontally aligned. It will first display all fields in the order defined in the form but skip special fields. Then it will display the special fields in the right way.

The actual implementation has some additional complexity that can help in certain situations to a look at the template files won’t hurt. This example is from templates/_form_functions.mako.

nl2br¶

The fluxscoreboard.util.nl2br() function can be used to turn newlines into line breaks for improved display in HTML. It’s usage is described on the API documentation

Now¶

To get the current time use datetime.datetime.utcnow() or time.time(). This yields UTC time (in the case of time this is irrelevant since the Epoch is timezone independent.

Converting Times¶

Conversion between a timestampe and a datetime object can be done with these functions (note that there is also the representation of timetuple which can also be helpful sometimes).

• timestamp –> datetime: datetime.datetime.utcfromtimestamp()
• datetime –> timestamp: datetime.datetime.utctimetuple() and calendar.timegm()

Displaying Times¶

The most easy way to display a time is to use the fluxscoreboard.util.tz_str() function:

from pytz import utc
from fluxscoreboard.util import tz_str
from datetime import datetime
local_time_str = tz_str(utc.localize(datetime.utcnow()),
                        request.team.timezone)

This is very easy because the team property is available in the request and it already know the teams timezone. Where you get the timestamp depends on what you are working on. For example, all timestamps in the database are managed in such a way that they already have the UTC timezone attached. Thus, for example, to display an announcements’ time, you would go like this:

from fluxscoreboard.util import tz_str
from fluxscoreboard.model.news import News
news = News()
local_time_str = tz_str(news.timestamp, request.team.timezone)

from pytz import utc, timezone
from datetime import datetime
timestamp = utc.localize(datetime.utcnow())
local_time = timestamp.astimezone(timezone("Europe/Berlin"))

New Timestamps¶

When you introduces new timestamps, always remember: UTC, always! If you want to create a specific time, attach a timezone upon creation:

from pytz import utc
from datetime import datetime
timestamp = datetime(2013, 1, 1, tzinfo=utc)

If you are saving it somewhere (i.e. database) save it as a naive UTC timestamp (saves you a lot of trouble) and then, upon retrieval, convert it to a timezone aware UTC timestamp. For example, take a look at fluxscoreboard.models.news.News which shows how you could implement this transparently.

Adding a New View¶

Views should be defined in the fluxscoreboard.views package. There you will find fluxscoreboard.views.front.BaseView from which you should derive all frontend views (as it provides access to some helpful attributes). You can also add your view to an existing view class which is most likely more what you want so let’s go down that road and define a new frontend view:

class FrontView(BaseView):
    ...

    @logged_in_view(route_name='my_route', renderer='my_route.mako')
    def my_route(self):
        # Do view stuff here
        return {}

Here’s what we did: We add a method to the class and decorate it with fluxscoreboard.views.front.logged_in_view. This decorator protects the view so only logged-in teams can see it. We assigned it the route name my_route and also specified a template my_route.mako. A template isn’t required, but usually only in a case where you redirect after you are with your view. After the view is done we return an empty dict. This contains the data passed to the template (in our case there is no data).

This was most of the work of adding a view. However, we still haven’t defined how to add a view. For this, you open up fluxscoreboard/routes.py and add a new route, like this:

routes = [ ...
          ('my_route', '/my_route'),
          ...
          ]

As you can see, adding a new route is easy: Now, whenever someone comes to the site and heads to /my_route, he will see whatever the view my_route wants to display.

Configuration¶

Behavior of the module can be changed by its configuration. These parameters all must be placed in a module-global dictionary called configuration and they should only relate to global configuration states, for everything else use the database and functions provided below.

allow_multiple is a boolean flag that determines whether more than one instance of this module is allowed. Most of the times you will want to set this to False.

Functions¶

The following functions can be defined to implement the dynamic behavior. They are all mandatory.

activate: Called during application startup and receives to parameters: config and settings, the first being the pyramid configurator instance and the second being the settings dictionary. You should register your routes here.

render: Called when the challenge should be rendered. Return everything you want to display for this challenge. Pay close attention to cover all the available states, e.g. archive mode, ctf has ended, etc. Look at the default challenge template for all important cases. Gets two parameters: The challenge represents the challenge instance this module belongs to, request is the current pyramid request object.

get_ponints: Return an integer denoting the number of points awarded for this challenge. Receives a team object representing the current team.

get_points_query: Receive an optional parameter on which class to work on, by default the Team table should be taken. Return a scalar subquery that fetches the current team’s points from the database. It should yield the same result as the get_points function except that this result will be calculated in the database.

title: Return a unicode string that denotes the title of the module to be displayed in the backend, i.e. something that identifies your module. Not shown in the frontend.

install: Install the module, performing any action that is required only on application installation.

Models & Views¶

Models and views are handled the same way as in the rest of the application: Use the Base class and view_config decorator. To register a view you can also work inside activate.

Templates¶

Put them in a directory called dynamic in the templates folder. You can also create a subdirectory there if you want.

Example¶

This is the (old, removed) flag challenge. It is not exactly up to date and does not completely use the interface specified above but an old version, but you get the idea.

# encoding: utf-8
from __future__ import unicode_literals, print_function, absolute_import
from fluxscoreboard.config import ROOT_DIR
from fluxscoreboard.models import Base, DBSession
from fluxscoreboard.models.challenge import Challenge
from fluxscoreboard.models.team import get_team_by_ref
from fluxscoreboard.util import now
from fluxscoreboard.views.front import BaseView
from pyramid.httpexceptions import HTTPFound
from pyramid.renderers import render
from pyramid.view import view_config
from requests.exceptions import RequestException
from sqlalchemy.orm import relationship, backref
from sqlalchemy.orm.exc import NoResultFound, MultipleResultsFound
from sqlalchemy.orm.mapper import validates
from sqlalchemy.schema import ForeignKey, Column
from sqlalchemy.sql.expression import func
from sqlalchemy.types import Integer, Unicode, BigInteger
from tempfile import mkdtemp
import csv
import logging
import os.path
import requests
import shutil
import socket
import transaction
import zipfile


log = logging.getLogger(__name__)
allow_multiple = False
"""Whether multiple instances of this are allowed"""
# GeoIP database
db_url = 'http://geolite.maxmind.com/download/geoip/database/GeoIPCountryCSV.zip'


class FlagView(BaseView):
    @view_config(route_name='ref', renderer="json")
    def ref(self):
        """
        Public view that is invoked at the ``ref`` route to which a client
        delivers a ``ref_id`` by which a team is found. This ID is then used
        to find the team it belongs to and upgrade its flag count if the
        location the client was from was not already in the list of registered
        locations for the team.
        """
        if self.archive_mode:
            flash_msg = "This challenge cannot be solved in archive mode."
            self.request.session.flash(flash_msg, 'error')
            return HTTPFound(location=self.request.route_url('home'))
        try:
            challenge = (DBSession.query(Challenge).
                         filter(Challenge.module == 'flags').one())
        except NoResultFound:
            ret = {'success': False, 'msg': ("There is no challenge for flags "
                                             "right now")}
            return ret
        except MultipleResultsFound:
            ret = {'success': False, 'msg': ("More than one challenge is "
                                             "online. This shouldn't happen, "
                                             "contact FluxFingers.")}
            return ret
        if (not challenge.online or
                self.request.settings.submission_disabled or
                now() > self.request.settings.ctf_end_date):
            ret = {'success': False}
            if not challenge.online:
                ret["msg"] = "Challenge is offline."
            elif self.request.settings.submission_disabled:
                ret["msg"] = "Submission is disabled."
            elif now() > self.request.settings.ctf_end_date:
                ret["msg"] = "CTF is over."
            return ret
        ref_id = self.request.matchdict["ref_id"]
        try:
            team = get_team_by_ref(ref_id)
        except NoResultFound:
            ret = {'success': False,
                   'msg': "Team not found."}
            return ret
        loc = get_location(self.request.client_addr)
        ret = {'success': True}
        if loc is None:
            log.warn("No valid location returned for IP address '%s' for "
                     "team '%s' with ref id '%s'"
                     % (self.request.client_addr, team, ref_id))
            ret["success"] = False
            ret["msg"] = ("No location found. Try a different IP from that "
                            "range.")
            return ret
        ret["location"] = loc
        try:
            t = transaction.savepoint()
            team.flags.append(loc)
            DBSession.flush()
        except Exception:
            ret["msg"] = "Location already registered."
            t.rollback()
        else:
            ret["msg"] = "Location successfully registered."
        return ret


class TeamFlag(Base):
    """
    Represent a quasi-many-to-many relationship between teams and flags. But
    the flags table is only present as a module-global variable and not in the
    database as it can be considered static (see :func:`install` for possible
    caveats).

    Recommended access to this is just going through a teams ``flags``
    attribute as it directly represents the flags already solves as a list of
    strings.

    .. todo::
        Once list is turned into a set of strings, update this documentation
        accordingly.
    """
    __tablename__ = 'team_flag'
    team_id = Column(Integer, ForeignKey('team.id'), primary_key=True)
    flag = Column(Unicode(2), primary_key=True)
    team = relationship("Team",
                        backref=backref("team_flags",
                                        cascade="all, delete-orphan"))

    def __init__(self, flag, **kwargs):
        kwargs["flag"] = flag
        Base.__init__(self, **kwargs)


class GeoIP(Base):
    """
    A mapping of an IP range to country codes. IP ranges are integers as they
    are natively anyway (4 blocks of 8 bit) and are stored this way for easier
    comparison.
    """
    ip_range_start = Column(BigInteger, primary_key=True,
                            autoincrement=False)
    ip_range_end = Column(BigInteger, nullable=False, unique=True, index=True)
    country_code = Column(Unicode(2), nullable=False)

    @staticmethod
    def ip_str(int_ip):
        """
        Turn an IP integer (such as those stored in the database) into a string
        for easier human-readable representation.
        """
        hex_ = hex(int_ip)[2:]
        if hex_.endswith("L"):
            hex_ = hex_[:-1]
        return socket.inet_ntoa(hex_.zfill(8).decode("hex"))

    @staticmethod
    def ip_int(str_ip):
        """
        Turn a human-readable string IP addressinto an integer IP address.
        """
        return int(socket.inet_aton(str_ip).encode("hex"), 16)

    @validates('ip_range_start', 'ip_range_end')
    def check_ip_range(self, key, ip):
        assert ip <= 0xFFFFFFFF
        assert ip >= 0
        return ip


def display(challenge, request):
    """
    Render the output for the challenge view. Displays a description and a
    grid of flags that can be visited.
    """
    from fluxscoreboard.models.team import get_team
    flags = []
    team = get_team(request)
    solved_flags = 0
    team_flags = set(team.flags) if team else set()
    for row in xrange(15):
        flag_row = []
        for col in xrange(15):
            index = row * 15 + col
            if index < len(flag_list):
                flag = flag_list[index]
                visited = flag in team_flags
                if visited:
                    solved_flags += 1
                flag_row.append((flag, visited))
        flags.append(flag_row)
    params = {'challenge': challenge,
              'flags': flags,
              'flag_stats': (solved_flags, len(flag_list)),
              'team': team}
    return render('dynamic_flags.mako', params, request)


def points_query(cls=None):
    """
    Returns a scalar query element that can be used in a ``SELECT`` statement
    to be added to the points query. The parameter ``cls`` can be anything
    that SQLAlchemy can correlate on. If left empty, it defaults to the
    standard :cls`fluxscoreboard.models.team.Team`, which is normally fine.
    However, if multiple teams are involved (as with the ranking algorithm)
    one might pass in an alias like this:

    .. code-block:: python
        inner_team = aliased(Team)
        dynamic_points = flags.points_query(inner_team)

    This will then correlate on a specific alias of ``Team`` instead of the
    default class.
    """
    if cls is None:
        from fluxscoreboard.models.team import Team
        cls = Team
    subquery = (DBSession.query(func.count('*')).
                filter(TeamFlag.team_id == cls.id).
                correlate(cls))
    return func.coalesce(subquery.as_scalar(), 0)


def points(team):
    return len(team.flags)


def get_location(ip):
    query = (DBSession.query(GeoIP.country_code).
             filter(GeoIP.ip_range_start <= GeoIP.ip_int(ip)).
             filter(GeoIP.ip_range_end >= GeoIP.ip_int(ip)))
    country_code, = query.first() or ("",)
    if country_code not in flag_list:
        log.info("Retrieved invalid country code '%s' for IP address %s. "
                 % (country_code, ip))
        return None
    else:
        return country_code


def title():
    return "Geolocation Flags (%s)" % __name__


def install(connection, with_update=True):
    geoip_fname = 'GeoIPCountryWhois.csv'
    geoip_file = os.path.join(ROOT_DIR, 'data', geoip_fname)
    if with_update:
        try:
            r = requests.get(db_url)
        except RequestException as e:
            log.error("Could not download current database because requests "
                      "threw an exception. This only means that the database will "
                      "not be up to date but we will use the old cached version. "
                      "Requests reported the following: '%s'" % e)
        else:
            tmpdir = mkdtemp()
            zipname = os.path.join(tmpdir, os.path.basename(db_url))
            with open(zipname, "w") as f:
                f.write(r.content)
            zip_ = zipfile.ZipFile(zipname)
            zip_.extractall(tmpdir)
            extracted_csv = os.path.join(tmpdir, geoip_fname)
            shutil.move(extracted_csv, geoip_file)
            shutil.rmtree(tmpdir)
    data = []
    available_country_codes = set()
    with open(geoip_file) as f:
        csv_ = csv.reader(f)
        for row in csv_:
            ip_int_start = int(row[2])
            ip_int_end = int(row[3])
            country_code = unicode(row[4].lower())
            if country_code not in flag_list:
                if country_code in flag_exceptions:
                    # Don't import it
                    continue
                else:
                    raise ValueError("The country code '%s' is not in the "
                                     "list of flags. It has the following "
                                     "data attached: '%s'"
                                     % (country_code, row))
            available_country_codes.add(country_code)
            item = {'ip_range_start': ip_int_start,
                    'ip_range_end': ip_int_end,
                    'country_code': country_code}
            data.append(item)
    log.info("Adding %d rows to database" % len(data))
    dialect = connection.dialect.name
    if dialect == "sqlite":
        chunk_size = 300
    elif dialect == "mysql":
        chunk_size = 10000
    else:
        chunk_size = len(data)

    while data:
        connection.execute(GeoIP.__table__.insert().values(data[:chunk_size]))
        data = data[chunk_size:]
    unreachable_countries = set(flag_list) - available_country_codes
    if unreachable_countries:
        log.warning("There are a number of countries that will not be "
                    "reachable for the teams because it is not present in our "
                    "database even though we display their flag. These "
                    "are the country codes that cannot be reached: '%s'"
                    % list(unreachable_countries))


flag_list = ['ad', 'ae', 'af', 'ag', 'ai', 'al', 'am', 'ao', 'aq',
             'ar', 'as', 'at', 'au', 'aw', 'az', 'ba', 'bb', 'bd', 'be',
             'bf', 'bg', 'bh', 'bi', 'bj', 'bm', 'bn', 'bo', 'br', 'bs',
             'bt', 'bw', 'by', 'bz', 'ca', 'cg', 'cf', 'cd', 'ch', 'ci',
             'ck', 'cl', 'cm', 'cn', 'co', 'cr', 'cu', 'cv', 'cy', 'cz',
             'de', 'dj', 'dk', 'dm', 'do', 'dz', 'ec', 'ee', 'eg', 'eh',
             'er', 'es', 'et', 'fi', 'fj', 'fm', 'fo', 'fr', 'ga', 'gb',
             'gd', 'ge', 'gg', 'gh', 'gi', 'gl', 'gm', 'gn', 'gp', 'gq',
             'gr', 'gt', 'gu', 'gw', 'gy', 'hk', 'hn', 'hr', 'ht', 'hu',
             'id', 'mc', 'ie', 'il', 'im', 'in', 'iq', 'ir', 'is', 'it',
             'je', 'jm', 'jo', 'jp', 'ke', 'kg', 'kh', 'ki', 'km', 'kn',
             'kp', 'kr', 'kw', 'ky', 'kz', 'la', 'lb', 'lc', 'li', 'lk',
             'lr', 'ls', 'lt', 'lu', 'lv', 'ly', 'ma', 'md', 'me', 'mg',
             'mh', 'mk', 'ml', 'mm', 'mn', 'mo', 'mq', 'mr', 'ms', 'mt',
             'mu', 'mv', 'mw', 'mx', 'my', 'mz', 'na', 'nc', 'ne', 'ng',
             'ni', 'nl', 'no', 'np', 'nr', 'nz', 'om', 'pa', 'pe', 'pf',
             'pg', 'ph', 'pk', 'pl', 'pr', 'ps', 'pt', 'pw', 'py', 'qa',
             're', 'ro', 'rs', 'ru', 'rw', 'sa', 'sb', 'sc', 'sd', 'se',
             'sg', 'si', 'sk', 'sl', 'sm', 'sn', 'so', 'sr', 'st', 'sv',
             'sy', 'sz', 'tc', 'td', 'tg', 'th', 'tj', 'tl', 'tm', 'tn',
             'to', 'tr', 'tt', 'tv', 'tw', 'tz', 'ua', 'ug', 'us', 'uy',
             'uz', 'va', 'vc', 've', 'vg', 'vi', 'vn', 'vu', 'ws', 'ye',
             'za', 'zm', 'zw']


# These are flags that exist in the original database but we do not recognize
# them
flag_exceptions = set(['eu', 'a2', 'yt', 'ap', 'tk', 'wf', 'cw', 'ss', 'a1',
                       'sh', 'cx', 'mf', 'gs', 'gf', 'cc', 'bl', 'nf', 'um',
                       'sj', 'bq', 'sx', 'mp', 'io', 'tf', 'ax', 'fk', 'pn',
                       'nu', 'pm'])

Transaction Support¶

The mailing system support transactions and thus makes sure that mails are only sent when the transaction succeeds. Thus, when an exception occurs, the mail is not sent. This provides a high level of integration and allows us to send mails thoughtlessly but be sure they are never sent until the actual application succeeded. However, this has a small performance impact on the response time: The email is only dispatched once the application completes and that means it cannot be deferred in any way: The application will only return once the email is sent. This results in a small noticeable delay.

Challenge Models & Functions¶

class fluxscoreboard.models.challenge.Challenge(**kwargs)[source]¶

A challenge in the system.

Attributes:

id: The primary key column.

title: Title of the challenge.

text: A description of the challenge.

solution: The challenge’s solution

points: How many points the challenge is worth.

online: Whether the challenge is online.

manual: If the points for this challenge are awareded manually.

category_id: ID of the associated category.

category: Direct access to the Category.

author: A simple string that contains an author (or a list thereof).

dynamic: Whether this challenge is dynamically handled. At the default of False this is just a normal challenge, otherwise, the attribute module must be set.

module: If this challenge is dynamic, it must provide a valid dotted python name for a module that provides the interface for validation and display. The dotted python name given here will be prefixed with fluxscoreboard.dynamic_challenges. from which the module will be loaded and made available on using it.

module: Loads the module from the module name and returns it.

published: Whether the challenge should be displayed in the frontend at all.

points[source]¶

The points of a challenge which is either the value assigned to it or, if the challenge is manual, the manual_challenge_points object to indicate that the points are manually assigned.

class fluxscoreboard.models.challenge.Category(**kwargs)[source]¶

A category for challenges.

Attributes:

id: Primary key of category.

name: Name of the category.

challenges: List of challenges in that category.

class fluxscoreboard.models.challenge.Submission(**kwargs)[source]¶

A single submission. Each entry means that this team has solved the corresponding challenge, i.e. there is no solved flag: The existence of the entry states that.

Attributes:

team_id: Foreign primary key column of the team.

challenge_id: Foreign primary key column of the challenge.

timestamp: A UTC-aware datetime.datetime object. When assigning a value always pass either a timezone-aware object or a naive UTC datetime. Defaults to datetime.datetime.utcnow().

bonus: How many bonus points were awared.

team: Direct access to the team who solved this challenge.

challenge: Direct access to the challenge.

class fluxscoreboard.models.challenge.Feedback(**kwargs)[source]¶

fluxscoreboard.models.challenge.get_all_challenges()[source]¶

Return a query that gets all challenges.

fluxscoreboard.models.challenge.get_online_challenges()[source]¶

Return a query that gets only those challenges that are online.

fluxscoreboard.models.challenge.get_submissions()[source]¶

Creates a query to eagerly load all submissions. That is, all teams and challenges that are attached to the submissions are fetched with them.

fluxscoreboard.models.challenge.get_all_categories()[source]¶

Get a list of all available categories.

fluxscoreboard.models.challenge.check_submission(challenge, solution, team, settings)[source]¶

Check a solution for a challenge submitted by a team and add it to the database if it was correct.

Args:

challenge: An instance of Challenge, the challenge to check the solution for.

solution: A string, the proposed solution for the challenge.

team: Team that submitted the solution.

Returns:

A tuple of (result, msg). result indicates whether the solution was accpeted (and added to the database) or not. The message returns a string with either a result (if result == False) or a congratulations message.

fluxscoreboard.models.challenge.manual_challenge_points = <ManualChallengePoints instance>¶

A static value that is returned instead of an actual number of points.

fluxscoreboard.models.challenge.update_playing_teams(connection)[source]¶

Update the number of playing teams whenever it changes.

fluxscoreboard.models.challenge.update_challenge_points(connection, update_team_count=True)[source]¶

Update the points on each challenge to reflect their current worth.

Country Models & Functions¶

class fluxscoreboard.models.country.Country(**kwargs)[source]¶

A country in the database. Basically only a name for different locations of teams.

fluxscoreboard.models.country.get_all_countries()[source]¶

Get a query that fetches a list of all countries from the database.

News Models & Functions¶

class fluxscoreboard.models.news.News(**kwargs)[source]¶

A single announcement, either global or for a challenge, depending on the challenge_id attribute.

Attributes:

id: The primary key.

timestamp: A UTC-aware datetime.datetime object. When assigning a value always pass either a timezone-aware object or a naive UTC datetime. Defaults to datetime.datetime.utcnow().

message: The text of the announcement.

published: Whether the announcement is displayed in the frontend.

challenge_id: If present, which challenge this announcement belongs to.

challenge: Direct access to the challenge, if any.

class fluxscoreboard.models.news.MassMail(**kwargs)[source]¶

An entry of a mass mail that was sent.

Attributes:

id: The primary key.

timestamp: A UTC-aware datetime.datetime object. When assigning a value always pass either a timezone-aware object or a naive UTC datetime. Defaults to datetime.datetime.utcnow().

subject: The subject of the mail

message: The body of the mail

recipients: A list of recipients that have recieved this mail. Internally this is stored as a json encoded list.

from_: The address which was used as the From: field of the mail.

fluxscoreboard.models.news.get_published_news()[source]¶

Team Models & Functions¶

class fluxscoreboard.models.team.Team(**kwargs)[source]¶

A team represented in the database.

Attributes:

id: Primary key

name: The name of the team.

password: The password of the team. If setting the password, pass it as cleartext. It will automatically be encrypted and stored in the database.

email: E-Mail address of the team. Verified if team is active.

country_id: Foreign Key specifying the location of the team.

local: Whether the team is local at the conference.

token: Token for E-Mail verification.

reset_token: When requesting a new password, this token is used.

challenge_token: Unique token for each team they can provide to

a challenge so this challenge can do rate-limiting or banning or whatever it wants to do.

active: Whether the team’s mail address has been verified and the team can actively log in.

timezone: A timezone, specified as a string, like "Europe/Berlin" or something that, when coerced to unicode, turns out as a string like this. Must be valid timezone.

acatar_filename: The filename under which the avatar is stored in the static/images/avatars directory.

size: The size of the team.

country: Direct access to the teams fluxscoreboard.models.country.Country attribute.

get_solvable_challenges()[source]¶

Return a list of challenges that the team can solve right now. It returns a list of challenges that are

• online
• unsolved by the current team
• not manual or dynamic (i.e. solvable by entering a solution)

get_unsolved_challenges()[source]¶

Return a query that produces a list of all unsolved challenges for a given team.

validate_password(password)[source]¶

Validate the password agains the team. If it matches return True else return False.

fluxscoreboard.models.team.get_all_teams()[source]¶

Get a query that returns a list of all teams.

fluxscoreboard.models.team.get_active_teams()[source]¶

Get a query that returns a list of all active teams.

fluxscoreboard.models.team.get_team_solved_subquery(team_id)[source]¶

Get a query that searches for a submission from a team for a given challenge. The challenge is supposed to come from an outer query.

Example usage:

team_solved_subquery = get_team_solved_subquery(team_id)
challenge_query = (DBSession.query(Challenge,
                                   team_solved_subquery))

In this example we query for a list of all challenges and additionally fetch whether the currenttly logged in team has solved it.

fluxscoreboard.models.team.get_number_solved_subquery()[source]¶

Get a subquery that returns how many teams have solved a challenge.

Example usage:

number_of_solved_subquery = get_number_solved_subquery()
challenge_query = (DBSession.query(Challenge,
                                   number_of_solved_subquery)

Here we query for a list of all challenges and additionally fetch the number of times it has been solved. This subquery will use the outer challenge to correlate on, so make sure to provide one or this query makes no sense.

fluxscoreboard.models.team.get_team(request)[source]¶

Get the currently logged in team. Returns None if the team is invalid (e.g. inactive) or noone is logged in or if the scoreboard is in archive mode.

fluxscoreboard.models.team.get_team_by_id(team_id)[source]¶

fluxscoreboard.models.team.register_team(form, request)[source]¶

Create a new team from a form and send a confirmation email.

Args:

form: A filled out fluxscoreboard.forms.front.RegisterForm.

request: The corresponding request.

Returns:

The Team that was created.

fluxscoreboard.models.team.send_activation_mail(team, request)[source]¶

Send activation mail to particular team.

fluxscoreboard.models.team.confirm_registration(token)[source]¶

For a token, check the database for the corresponding team and activate it if found.

Args:

token: The token that was sent to the user (a string)

Returns:

Either True or False depending on whether the confirmation was successful.

fluxscoreboard.models.team.login(email, password)[source]¶

Check a combination of credentials for validaity and either return a reason why it failed or return the logged in team.

Args:

email: The email address of the team.

password: The corresponding password.

Returns:

A three-tuple of (result, message, team). result indicates whether the login was successful or not. In case of failure msg contains a reason why it failed so it can be logged (but not printed - we don’t want to give any angle to an attacker). If the login was successful, msg is None. Finally, if the login succeeded, team contains the found instance of Team. If login failed, team is None.

fluxscoreboard.models.team.password_reminder(email, request)[source]¶

For an email address, find the corresponding team and send a password reset token. If no team is found send an email that no user was found for this address.

fluxscoreboard.models.team.check_password_reset_token(token)[source]¶

Check if an entered password reset token actually exists in the database.

fluxscoreboard.models.team.TEAM_GROUPS = [u'group:team']¶

Groups are just fixed: If a team is logged in it belongs to these groups.

fluxscoreboard.models.team.groupfinder(userid, request)[source]¶

Check if there is a team logged in, and if it is, return the default TEAM_GROUPS.

class fluxscoreboard.models.team.TeamIP(**kwargs)[source]¶

fluxscoreboard.models.team.update_score(connection, update_all=True)[source]¶

Update the score of all teams. If update_all is set, the points for all challenges are updated beforehand as well.

This is your one-shot function to create up-to-date points for everything.

Settings Models & Functions¶

class fluxscoreboard.models.settings.Settings(**kwargs)[source]¶

Represents application settings. Do not insert rows of this. There must always be only one row which is updated. This is preferred to having multiple rows with only key->value because with this we can enforce types and give an overview over available settings.

The most straightforward usage is calling get() to retrieve the current settings which you can then also edit and they will be saved automatically.

The following settings are available:

submission_disabled: A boolean that describes whether currently submissions are allowed or disabled. Default: False

ctf_start_date: A timezone-aware datetime value that describes when the CTF should start. Before that, the application will behave differently, e.g. may not allow login.

ctf_end_date: When the CTF will end. Same type as ctf_start_date.

ctf_started: This is a property that can only be read and just compares ctf_start_date with the current time to find out whether the CTF has already started or not.

archive_mode: When the scoreboard is in archive mode, the frontend will not allow alteration to the database. Additionally, the whole system is public so everyone can get their solutions checked. This is then verified and the result is returned, but it is not added to the database. The following things will change in archive mode:

• No registration
• No login
• Start / End times ignored
• Solutions can be submitted but will only return the result, not enter something into the databse
• Challenges are public in addition to the scoreboard

ctf_state: Which time state the CTF currently is in. Relevant for

permissions etc.

Custom Column Types¶

class fluxscoreboard.models.types.TZDateTime(*args, **kwargs)[source]¶

Coerces a tz-aware datetime object into a naive utc datetime object to be stored in the database. If already naive, will keep it.

On return of the data will restore it as an aware object by assuming it is UTC.

Use this instead of the standard sqlalchemy.types.DateTime.

class fluxscoreboard.models.types.Timezone(*args, **kwargs)[source]¶

Represent a timezone, storing it as a unicode string but giving back a datetime.tzinfo instance.

class fluxscoreboard.models.types.Module(*args, **kwargs)[source]¶

Represent a python module from the dynamic_challenges submodule. Input is a string but the return value will always be a real python module.

Custom Forms¶

class fluxscoreboard.forms.CSRFForm(formdata=None, obj=None, prefix=u'', csrf_context=None, **kwargs)[source]¶

Todo

Document.

Frontend Forms¶

This module contains all forms for the frontend. Backend forms can be found in fluxscoreboard.forms.admin. The forms here and there are described as they should behave, e.g. “Save the challenge”, however, this behaviour has to be implemented by the developer and is not done automatically by it. However, validatation restrictions (e.g. length) are enforced. But alone, without a database to persist them, they are mostly useless.

class fluxscoreboard.forms.front.RegisterForm(formdata=None, obj=None, prefix=u'', csrf_context=None, **kwargs)[source]¶

Registration form for new teams.

Attrs:

name: The name of the team, required and must have a length between 1 and 255.

email: The teams E-Mail address, required, must be the same as email_repeat, must have a length between 5 and 255. It is also checked that this email is not registered yet.

email_repeat: Must have the same value as email.

password: The team’s password, must be between 8 and 1024 characters long.

password_repeat: Must have the same value as password.

country: A list of available countries in the database to choose from.

timezone: The timezone to apply for the team. When not selected, UTC is used, otherwise this localizes times displayed on the frontpage.

submit: The submit button.

class fluxscoreboard.forms.front.LoginForm(formdata=None, obj=None, prefix=u'', csrf_context=None, **kwargs)[source]¶

Login form for teams that are activated.

Attrs:

email: Login email address.

password: Login password.

login: Submit button.

class fluxscoreboard.forms.front.ForgotPasswordForm(formdata=None, obj=None, prefix=u'', csrf_context=None, **kwargs)[source]¶

A form to get an email for a forgotten password.

class fluxscoreboard.forms.front.ResetPasswordForm(formdata=None, obj=None, prefix=u'', csrf_context=None, **kwargs)[source]¶

A form to finish a started reset process.

class fluxscoreboard.forms.front.ProfileForm(formdata=None, obj=None, prefix=u'', csrf_context=None, **kwargs)[source]¶

A form to edit a team’s profile.

Attrs:

email: The email address. Required

old_password: The old password, needed only for a password change.

password: The password. Optional, only needed if wanting to change.

password_repeat: Repeat the new password.

avatar: Display an avatar and upload a new one.

country: Change location. Required.

timezone: Change timezone. Required.

submit: Save changes

cancel: Abort

class fluxscoreboard.forms.front.SolutionSubmitForm(formdata=None, obj=None, prefix=u'', csrf_context=None, **kwargs)[source]¶

A form to submit a solution for a challenge on a single challenge view. This form does not keep track of which challenge this is.

Attrs:

solution: The proposed solution for the challenge. Required.

submit: Submit the solution.

class fluxscoreboard.forms.front.SolutionSubmitListForm(formdata=None, obj=None, prefix=u'', csrf_context=None, **kwargs)[source]¶

A form to submit a solution for any challenge selected from a list. Keeps track of which challenge the solution was submitted for. Subclass of SolutionSubmitForm and derives all attributes from it. New attributes defined here:

challenge: A list of challenges to choose from. Required.

class fluxscoreboard.forms.front.FeedbackForm(formdata=None, obj=None, prefix=u'', csrf_context=None, **kwargs)[source]¶

Utility Functions¶

fluxscoreboard.util.display_design(request)[source]¶

A function that returns True or False depending on whether the actual design should be displayed (True) or just a default one (False).

This is used to not disclose the real design until the CTF has started. The following logic is implemented:

• The admin backend has the default design (i.e. False).
• If it is a test-login from the backend, show the design.
• If the CTF has started, the real design is loaded for the frontpage (i.e. True).
• If no route is matched, the default design is loaded (False).
• Otherwise we fall back to not show the design (False).

The conditions are processed in order, i.e. the first match is returned.

fluxscoreboard.util.is_admin_path(request)[source]¶

fluxscoreboard.util.now()[source]¶

Return the current timestamp localized to UTC.

fluxscoreboard.util.random_token(length=64)[source]¶

Generate a random token hex-string of length characters. Due to it being encoded hex, its entropy is only half, so if you require a 256 bit entropy string, passing in 64 as length will yield exactly 64 / 2 * 8 = 256 bits entropy.

fluxscoreboard.util.nl2br(text)[source]¶

Translate newlines into HTML <br /> tags.

Usage:

<% from fluxscoreboard.util import nl2br %>
${some_string | nl2br}

fluxscoreboard.util.random_str(len_, choice='abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ')[source]¶

fluxscoreboard.util.tz_str(timestamp, timezone=None)[source]¶

Create a localized timestring in a local timezone from the timezone-aware datetime.datetime object timestamp.

class fluxscoreboard.util.not_logged_in(msg=None)[source]¶

Decorator for a view that should only be visible to users that are not logged in. They will be redirected to the frontpage and a message will be shown, that can be specified, but also has a sensible default.

Usage:

@view_config(...)
@not_logged_in()
def some_view(request):
    pass

Route Definition¶

All routes of the application in a list. Each item consists of the route name and its path in a tuple: (route_name, path). These are then mapped to a view via the pyramid.view.view_config decorator. See Routing and Views for an explanation of how to create routes and work with them.

Installation Utilites¶

Installation module that provides the application with mechanisms for installing and uninstalling the application. Also useful for testing.

fluxscoreboard.install.install(settings, cfg_file, test_data=False)[source]¶

Installs the application. Only required to be called once per installation.

fluxscoreboard.install.uninstall(settings)[source]¶

Remove those parts created by install