Skip to content

Commit 29012ee

Browse files
gsiri-codegsiri-code
committed
feat(security): add protocol manager and opt-in enforcement
1 parent bd5e408 commit 29012ee

5 files changed

Lines changed: 327 additions & 0 deletions

File tree

‎security/__init__.py‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1 +1,12 @@
1+
from security.decorators import require_protocol
2+
from security.manager import get_protocol, is_permitted, load_legacy_records
3+
from security.models import ActionChecks, SecProtocol
14

5+
__all__ = [
6+
"ActionChecks",
7+
"SecProtocol",
8+
"get_protocol",
9+
"is_permitted",
10+
"load_legacy_records",
11+
"require_protocol",
12+
]

‎security/decorators.py‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
"""Opt-in route security helpers.
2+
3+
Enforcement is disabled by default to preserve current API behavior.
4+
"""
5+
6+
from __future__ import annotations
7+
8+
import logging
9+
import os
10+
from functools import wraps
11+
from http import HTTPStatus
12+
13+
from flask import g, request
14+
15+
from security.manager import is_permitted
16+
17+
LOGGER = logging.getLogger(__name__)
18+
19+
20+
def security_enforcement_enabled() -> bool:
21+
return os.getenv("SECURITY_ENFORCEMENT", "false").lower() in {"1", "true", "yes", "on"}
22+
23+
24+
def security_audit_only() -> bool:
25+
return os.getenv("SECURITY_AUDIT_ONLY", "false").lower() in {"1", "true", "yes", "on"}
26+
27+
28+
def require_protocol(feature_name: str, action: str):
29+
def decorator(fn):
30+
@wraps(fn)
31+
def wrapper(*args, **kwargs):
32+
permitted = is_permitted(
33+
feature_name,
34+
action,
35+
auth_header=request.headers.get("Authorization"),
36+
)
37+
g.security_result = {
38+
"feature": feature_name,
39+
"action": action,
40+
"permitted": permitted,
41+
}
42+
if not security_enforcement_enabled():
43+
return fn(*args, **kwargs)
44+
if security_audit_only():
45+
LOGGER.info("Security audit result: %s", g.security_result)
46+
return fn(*args, **kwargs)
47+
if not permitted:
48+
return {"message": "forbidden"}, HTTPStatus.FORBIDDEN
49+
return fn(*args, **kwargs)
50+
51+
return wrapper
52+
53+
return decorator

‎security/manager.py‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
"""In-memory protocol manager for team-bemg security."""
2+
3+
from __future__ import annotations
4+
5+
from copy import deepcopy
6+
7+
from security.models import SecProtocol, protocol_from_legacy
8+
from server.auth import authenticate_request
9+
10+
protocols: dict[str, SecProtocol] = {}
11+
legacy_records: dict[str, dict] = {}
12+
13+
14+
def clear() -> None:
15+
protocols.clear()
16+
legacy_records.clear()
17+
18+
19+
def load_legacy_records(records: dict[str, dict]) -> dict[str, SecProtocol]:
20+
clear()
21+
legacy_records.update(deepcopy(records))
22+
for feature_name, feature_record in legacy_records.items():
23+
protocols[feature_name] = protocol_from_legacy(feature_name, feature_record)
24+
return protocols
25+
26+
27+
def add_protocol(protocol: SecProtocol) -> None:
28+
if protocol.name in protocols:
29+
raise ValueError(f"Duplicate protocol: {protocol.name}")
30+
protocols[protocol.name] = protocol
31+
32+
33+
def get_protocol(name: str) -> SecProtocol | None:
34+
return protocols.get(name)
35+
36+
37+
def exists(name: str) -> bool:
38+
return name in protocols
39+
40+
41+
def _auth_payload_from_inputs(
42+
auth_header: str | None = None,
43+
auth_payload: dict | None = None,
44+
) -> dict:
45+
if auth_payload is not None:
46+
return auth_payload
47+
if not auth_header:
48+
return {}
49+
try:
50+
return authenticate_request(auth_header)
51+
except PermissionError:
52+
return {}
53+
54+
55+
def is_permitted(
56+
feature_name: str,
57+
action: str,
58+
user_id: str = "",
59+
auth_header: str | None = None,
60+
auth_payload: dict | None = None,
61+
api_key: str = "",
62+
phrase: str = "",
63+
code: str | None = None,
64+
) -> bool:
65+
protocol = get_protocol(feature_name)
66+
if protocol is None:
67+
return True
68+
69+
payload = _auth_payload_from_inputs(auth_header=auth_header, auth_payload=auth_payload)
70+
effective_user_id = user_id or payload.get("sub", "")
71+
check_vals = {
72+
"role": payload.get("role"),
73+
"api_key": api_key,
74+
"phrase": phrase,
75+
"code": code,
76+
}
77+
return protocol.is_permitted(action, user_id=effective_user_id, check_vals=check_vals)

‎security/models.py‎

Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
"""Security protocol models with backward-compatible, additive checks."""
2+
3+
from __future__ import annotations
4+
5+
from copy import deepcopy
6+
7+
CREATE = "create"
8+
READ = "read"
9+
UPDATE = "update"
10+
DELETE = "delete"
11+
VALID_ACTIONS = [CREATE, READ, UPDATE, DELETE]
12+
13+
USER_LIST = "user_list"
14+
CHECKS = "checks"
15+
LOGIN = "login"
16+
ALLOWED_ROLES = "allowed_roles"
17+
API_KEY = "api_key"
18+
API_KEYS = "api_keys"
19+
PASS_PHRASE = "pass_phrase"
20+
PASSWORD = "password"
21+
CODES = "codes"
22+
23+
24+
class ActionChecks:
25+
"""Represents the checks required for one CRUD action."""
26+
27+
def __init__(
28+
self,
29+
login: bool = False,
30+
valid_users: list[str] | None = None,
31+
allowed_roles: list[str] | None = None,
32+
api_key: bool = False,
33+
valid_api_keys: list[str] | None = None,
34+
pass_phrase: bool = False,
35+
phrase: str = "",
36+
codes: dict[str, str] | None = None,
37+
):
38+
if not isinstance(login, bool):
39+
raise TypeError("login must be a bool")
40+
if not isinstance(api_key, bool):
41+
raise TypeError("api_key must be a bool")
42+
if not isinstance(pass_phrase, bool):
43+
raise TypeError("pass_phrase must be a bool")
44+
if valid_users is not None and not isinstance(valid_users, list):
45+
raise TypeError("valid_users must be a list or None")
46+
if allowed_roles is not None and not isinstance(allowed_roles, list):
47+
raise TypeError("allowed_roles must be a list or None")
48+
if valid_api_keys is not None and not isinstance(valid_api_keys, list):
49+
raise TypeError("valid_api_keys must be a list or None")
50+
if codes is not None and not isinstance(codes, dict):
51+
raise TypeError("codes must be a dict or None")
52+
53+
self.login = login
54+
self.valid_users = list(valid_users or [])
55+
self.allowed_roles = list(allowed_roles or [])
56+
self.api_key = api_key
57+
self.valid_api_keys = list(valid_api_keys or [])
58+
self.pass_phrase = pass_phrase
59+
self.phrase = phrase
60+
self.codes = deepcopy(codes) if codes else None
61+
62+
def to_json(self) -> dict:
63+
payload = {
64+
LOGIN: self.login,
65+
API_KEY: self.api_key,
66+
PASS_PHRASE: self.pass_phrase,
67+
}
68+
if self.valid_users:
69+
payload[USER_LIST] = list(self.valid_users)
70+
if self.allowed_roles:
71+
payload[ALLOWED_ROLES] = list(self.allowed_roles)
72+
if self.valid_api_keys:
73+
payload[API_KEYS] = list(self.valid_api_keys)
74+
if self.phrase:
75+
payload[PASSWORD] = self.phrase
76+
if self.codes:
77+
payload[CODES] = deepcopy(self.codes)
78+
return payload
79+
80+
def is_permitted(self, user_id: str = "", check_vals: dict | None = None) -> bool:
81+
check_vals = check_vals or {}
82+
if self.login and not user_id:
83+
return False
84+
if self.valid_users and user_id not in self.valid_users:
85+
return False
86+
if self.allowed_roles and check_vals.get("role") not in self.allowed_roles:
87+
return False
88+
if self.api_key and check_vals.get("api_key") not in self.valid_api_keys:
89+
return False
90+
if self.pass_phrase and check_vals.get("phrase") != self.phrase:
91+
return False
92+
if self.codes and check_vals.get("code") not in self.codes.values():
93+
return False
94+
return True
95+
96+
97+
class SecProtocol:
98+
"""Represents security rules for a named feature/resource."""
99+
100+
def __init__(
101+
self,
102+
name: str,
103+
create: ActionChecks | None = None,
104+
read: ActionChecks | None = None,
105+
update: ActionChecks | None = None,
106+
delete: ActionChecks | None = None,
107+
):
108+
if not isinstance(name, str):
109+
raise TypeError("name must be a str")
110+
self.name = name
111+
self.create = create or ActionChecks()
112+
self.read = read or ActionChecks()
113+
self.update = update or ActionChecks()
114+
self.delete = delete or ActionChecks()
115+
116+
def to_json(self) -> dict:
117+
return {
118+
"feature_name": self.name,
119+
CREATE: self.create.to_json(),
120+
READ: self.read.to_json(),
121+
UPDATE: self.update.to_json(),
122+
DELETE: self.delete.to_json(),
123+
}
124+
125+
def is_permitted(self, action: str, user_id: str = "", check_vals: dict | None = None) -> bool:
126+
if action not in VALID_ACTIONS:
127+
raise ValueError(f"Invalid action: {action}")
128+
return getattr(self, action).is_permitted(user_id=user_id, check_vals=check_vals)
129+
130+
131+
def checks_from_legacy(action_record: dict | None) -> ActionChecks:
132+
if not action_record:
133+
return ActionChecks()
134+
checks = action_record.get(CHECKS, {})
135+
return ActionChecks(
136+
login=checks.get(LOGIN, False),
137+
valid_users=action_record.get(USER_LIST, []),
138+
allowed_roles=checks.get(ALLOWED_ROLES, []),
139+
api_key=checks.get(API_KEY, False),
140+
valid_api_keys=checks.get(API_KEYS, []),
141+
pass_phrase=checks.get(PASS_PHRASE, False),
142+
phrase=checks.get(PASSWORD, ""),
143+
codes=checks.get(CODES),
144+
)
145+
146+
147+
def protocol_from_legacy(feature_name: str, feature_record: dict | None) -> SecProtocol:
148+
feature_record = feature_record or {}
149+
return SecProtocol(
150+
feature_name,
151+
create=checks_from_legacy(feature_record.get(CREATE)),
152+
read=checks_from_legacy(feature_record.get(READ)),
153+
update=checks_from_legacy(feature_record.get(UPDATE)),
154+
delete=checks_from_legacy(feature_record.get(DELETE)),
155+
)

‎security/security.py‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
from functools import wraps
22

3+
from security.manager import get_protocol, is_permitted as manager_is_permitted, load_legacy_records
4+
35
# import data.db_connect as dbc
46

57
"""
@@ -78,6 +80,7 @@ def read() -> dict:
7880
global security_recs
7981
# dbc.read()
8082
security_recs = temp_recs
83+
load_legacy_records(security_recs)
8184
return security_recs
8285

8386

@@ -100,3 +103,31 @@ def read_feature(feature_name: str) -> dict:
100103
return security_recs[feature_name]
101104
else:
102105
return None
106+
107+
108+
@needs_recs
109+
def read_protocol(feature_name: str):
110+
return get_protocol(feature_name)
111+
112+
113+
@needs_recs
114+
def is_permitted(
115+
feature_name: str,
116+
action: str,
117+
user_id: str = '',
118+
auth_header: str | None = None,
119+
auth_payload: dict | None = None,
120+
api_key: str = '',
121+
phrase: str = '',
122+
code: str | None = None,
123+
) -> bool:
124+
return manager_is_permitted(
125+
feature_name,
126+
action,
127+
user_id=user_id,
128+
auth_header=auth_header,
129+
auth_payload=auth_payload,
130+
api_key=api_key,
131+
phrase=phrase,
132+
code=code,
133+
)

0 commit comments

Comments
 (0)