Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/securitykey.py: 17%
66 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-25 15:02 +0000
1"""
2 Implementation of one-time CSRF-security-keys.
4 CSRF-security-keys (Cross-Site Request Forgery) are used mostly to make requests unique and non-reproducible.
5 Doing the same request again requires to obtain a fresh security key first.
6 Furthermore, security keys can be used to implemented credential-reset mechanisms or similar features, where a
7 URL is only valid for one call.
9 ..note:
10 There's also a hidden 3rd type of security-key: The session's static security key.
12 This key is only revealed once during login, as the protected header "Sec-X-ViUR-StaticSessionKey".
14 This can be used instead of the one-time sessions security key by sending it back as the same protected HTTP
15 header and setting the skey value to "STATIC_SESSION_KEY". This is only intended for non-web-browser,
16 programmatic access (admin tools, import tools etc.) where CSRF attacks are not applicable.
18 Therefor that header is prefixed with "Sec-" - so it cannot be read or set using JavaScript.
19"""
20import typing as t
21import datetime
22import hmac
23from viur.core import conf, utils, current, db, tasks
25SECURITYKEY_KINDNAME = "viur-securitykey"
26SECURITYKEY_DURATION = 24 * 60 * 60 # one day
27SECURITYKEY_STATIC_HEADER: t.Final[str] = "Sec-X-ViUR-StaticSessionKey"
28"""The name of the header in which the static session key is provided at login
29and must be specified in requests that require a skey."""
30SECURITYKEY_STATIC_SKEY: t.Final[str] = "STATIC_SESSION_KEY"
31"""Value that must be used as a marker in the payload (key: skey) to indicate
32that the session key from the headers should be used."""
35def create(
36 duration: None | int | datetime.timedelta = None,
37 session_bound: bool = True,
38 key_length: int = 13,
39 indexed: bool = True,
40 amount: int = 1,
41 **custom_data,
42) -> str | tuple[str]:
43 """
44 Creates a new one-time CSRF-security-key.
46 The custom data (given as **custom_data) that can be stored with the key.
47 Any data provided must be serializable by the datastore.
49 :param duration: Make this CSRF-token valid for a fixed timeframe.
50 :param session_bound: Bind this CSRF-token to the current session.
51 :param indexed: Indexes all values stored with the security-key (default), set False to not index.
52 :param key_length: Allows to modify the length of the generated randomized key
53 :param custom_data: Any other data is stored with the CSRF-token, for later re-use.
54 :param amount: The amount of CSRF-tokens to generate.
56 :returns: The new one-time key. This is always a randomized string. \
57 In case amount > 1 is returned, it will be a tuple of strings with the keys.
58 """
59 if any(k.startswith("viur_") for k in custom_data):
60 raise ValueError("custom_data keys with a 'viur_'-prefix are reserved.")
61 if amount < 1 or amount > 500:
62 raise ValueError("amount must be between 1 and 500.")
63 if not duration:
64 duration = conf.user.session_life_time if session_bound else SECURITYKEY_DURATION
66 entities = []
67 for i in range(amount):
68 key = utils.string.random(key_length)
69 entity = db.Entity(db.Key(SECURITYKEY_KINDNAME, key))
70 entity |= custom_data
71 if session_bound:
72 session = current.session.get()
73 if not session.loaded:
74 session.reset()
75 entity["viur_session"] = session.cookie_key
77 else:
78 entity["viur_session"] = None
80 entity["viur_until"] = utils.utcNow() + utils.parse.timedelta(duration)
82 if not indexed:
83 entity.exclude_from_indexes = [k for k in entity.keys() if not k.startswith("viur_")]
84 entities.append(entity)
86 db.put(entities)
88 if amount > 1:
89 return tuple(entity.key.id_or_name for entity in entities)
90 return key
93def validate(key: str, session_bound: bool = True) -> bool | db.Entity:
94 """
95 Validates a CSRF-security-key.
97 :param key: The CSRF-token to be validated.
98 :param session_bound: If True, make sure the CSRF-token is created inside the current session.
99 :returns: False if the key was not valid for whatever reasons, the data (given during :meth:`create`) as
100 dictionary or True if the dict is empty (or session was True).
101 """
102 if session_bound and key == SECURITYKEY_STATIC_SKEY:
103 if skey_header_value := current.request.get().request.headers.get(SECURITYKEY_STATIC_HEADER):
104 return hmac.compare_digest(current.session.get().static_security_key, skey_header_value)
106 return False
108 if not key or not (entity := db.get(db.Key(SECURITYKEY_KINDNAME, key))):
109 return False
111 # First of all, delete the entity, validation is done afterward.
112 db.delete(entity)
114 # Key has expired?
115 if entity["viur_until"] < utils.utcNow():
116 return False
118 del entity["viur_until"]
120 # Key is session bound?
121 if session_bound:
122 if entity["viur_session"] != current.session.get().cookie_key:
123 return False
124 elif entity["viur_session"]:
125 return False
127 del entity["viur_session"]
129 return entity or True
132@tasks.PeriodicTask(interval=datetime.timedelta(hours=4))
133def periodic_clear_skeys():
134 from viur.core import tasks
135 """
136 Removes expired CSRF-security-keys periodically.
137 """
138 query = db.Query(SECURITYKEY_KINDNAME).filter("viur_until <", utils.utcNow() - datetime.timedelta(seconds=300))
139 tasks.DeleteEntitiesIter.startIterOnQuery(query)
142@tasks.CallDeferred
143def clear_session_skeys(session_key):
144 from viur.core import tasks
145 """
146 Removes any CSRF-security-keys bound to a specific session.
147 This function is called by the Session-module based on reset-actions.
148 """
149 query = db.Query(SECURITYKEY_KINDNAME).filter("viur_session", session_key)
150 tasks.DeleteEntitiesIter.startIterOnQuery(query)