the LAN rule moves from nginx into the app, as a setting for keys only

api_keys_from is a typed setting, lan (default, the historical rule) or
anywhere, checked in admin_api._auth against the first X-Forwarded-For hop
that NPM sets. It governs API keys only: a session is never restricted, the
console is a session from anywhere; and the break-glass token is ALWAYS
LAN-only, which is not a choice and so is not a setting. LAN ranges are
facts about the network and live in code; the docker range is included
because NPM, the routines and sibling containers reach the app from
arrstack_arr_net. An unparseable address is not LAN - the rule fails closed.

With this in place the location /api/admin/ block on NPM host 51 can come
out; the app enforces what it enforced, and the toggle never touches NPM.

tests/smoke_admin.py 58 -> 76. Proven on a throwaway site: key LAN 200,
key WAN 403 naming the setting, session WAN 200, token LAN 200, token WAN
403; set anywhere, key WAN 200 and token WAN still 403; bogus value 422.
This commit is contained in:
2026-09-04 17:45:17 -04:00
parent 4b5b2a3667
commit 69b177d519
3 changed files with 97 additions and 3 deletions
+46 -1
View File
@@ -159,6 +159,51 @@ check("leader cannot touch settings", not I.can(leader, "settings:write"))
check("admin spans units", I.can(admin, "unit:write_own", "u1") and I.can(admin, "unit:write_own", "u2"))
check("admin holds settings:write", I.can(admin, "settings:write"))
print("\nlan rule")
import admin_api as A
class _Req:
def __init__(self, ip, cookie=None, bearer=None):
self.headers = {"x-forwarded-for": ip} if ip else {}
if bearer: self.headers["authorization"] = "Bearer " + bearer
self.cookies = {"s73_session": cookie} if cookie else {}
self.client = None
for ip, want in (("10.0.0.55", True), ("10.0.1.20", True), ("127.0.0.1", True), ("172.19.0.31", True),
("108.36.248.87", False), ("192.168.1.9", False), ("", False), ("garbage", False)):
check("client_is_lan(%r) is %s" % (ip, want), A.client_is_lan(_Req(ip)) == want)
check("api_keys_from defaults to lan", I.get_setting("api_keys_from") == "lan")
raises("api_keys_from rejects other values", 422, I.IdentityError, I.set_setting, "api_keys_from", "vpn")
# a key from outside is refused while lan, allowed once anywhere, admin token never from outside
A.ADMIN_TOKEN = "t"
import uuid as _uuid
con = I.connect()
pid = str(_uuid.uuid4())
con.execute("INSERT INTO people (id, email, full_name, global_role, created_at) VALUES (?,?,?,?,?)",
(pid, "lanrule@example.test", "Lan Rule", "admin", I._now())); con.commit(); con.close()
person = I.get_person(pid)
full, _row = I.mint_api_key(person, "outside", ["leads:read"])
def _hits(req, cap="leads:read"):
try:
A._auth(req, None, cap); return 200
except Exception as e:
return getattr(e, "status_code", 500)
check("key from the LAN passes", _hits(_Req("10.0.0.55", bearer=full)) == 200)
check("key from outside is 403 while lan", _hits(_Req("108.36.248.87", bearer=full)) == 403)
I.set_setting("api_keys_from", "anywhere", actor="test")
check("key from outside passes once anywhere", _hits(_Req("108.36.248.87", bearer=full)) == 200)
check("key still cannot exceed its scopes from anywhere", _hits(_Req("108.36.248.87", bearer=full), "nearby:write") == 403)
I.set_setting("api_keys_from", None, actor="test")
check("clearing the setting restores lan", I.get_setting("api_keys_from") == "lan"
and _hits(_Req("108.36.248.87", bearer=full)) == 403)
tok = I.start_session(pid)
check("a session from outside is never restricted", _hits(_Req("108.36.248.87", cookie=tok)) == 200)
def _tok(req):
try:
A._auth(req, "t", "leads:read"); return 200
except Exception as e:
return getattr(e, "status_code", 500)
check("admin token from the LAN passes", _tok(_Req("10.0.0.55")) == 200)
check("admin token from outside is 403, regardless of the setting", _tok(_Req("108.36.248.87")) == 403)
print("\napi docs registry")
import admin_api as A
reg = A.describe_routes()
@@ -169,7 +214,7 @@ check("every route's capability is read from its own _auth call (except whoami):
check("key routes gate on apikeys:own", all(d["capability"] == "apikeys:own" for d in reg if "/keys" in d["path"]))
check("docs carry the handler docstring", all(d["doc"] for d in reg if "/keys" in d["path"]))
class _R:
headers = {"authorization": ""}; cookies = {}
headers = {"authorization": "", "x-forwarded-for": "127.0.0.1"}; cookies = {}; client = None
os.environ["ADMIN_TOKEN"] = "t"; A.ADMIN_TOKEN = "t"
page = A.api_docs(_R(), "t").body.decode()
check("docs page renders every route", page.count("<tr><td><code>") == len(reg))