redis-security/examples/probe-redis.py

Version a84871d065f3.bb1 · MIT. This preview displays packaged text and does not execute code. Treat the contents as untrusted instructions.

← Return to resource and package checksum

"""Synthetic Redis command observations. MIT, Copyright 2026 BB Skills."""
import socket,json,secrets,sys,datetime as dt,hashlib
from pathlib import Path
class Client:
 def __init__(self):self.s=socket.create_connection(('bb-v44-redis',6379),timeout=5);self.f=self.s.makefile('rb')
 def close(self):self.f.close();self.s.close()
 def read(self):
  line=self.f.readline()
  if not line:raise EOFError('connection-closed')
  kind=line[:1];value=line[1:-2]
  if kind==b'+':return value.decode()
  if kind==b'-':return {'error':value.decode().split(' ',1)[0]}
  if kind==b':':return int(value)
  if kind==b'$':
   n=int(value)
   if n==-1:return None
   data=self.f.read(n);assert self.f.read(2)==b'\r\n';return data.decode()
  if kind==b'*':
   n=int(value);return None if n==-1 else [self.read() for _ in range(n)]
  raise ValueError('Unsupported RESP2 response')
 def cmd(self,*args):
  values=[str(a).encode() for a in args];self.s.sendall(b'*'+str(len(values)).encode()+b'\r\n'+b''.join(b'$'+str(len(v)).encode()+b'\r\n'+v+b'\r\n' for v in values));return self.read()
cases=[];clients=[]
def client():c=Client();clients.append(c);return c
def check(name,observed,expected):
 row={'name':name,'observed':observed,'expected':expected,'status':'passed' if observed==expected else 'failed'};cases.append(row);assert observed==expected,row
def ttl(name,c,key):value=c.cmd('TTL',key);check(name,0<value<=30,True);return value
try:
 bootstrap=json.loads(sys.stdin.readline())['bootstrap_password']
 a=client();check('initial-default-authentication',a.cmd('AUTH','default',bootstrap),'OK');check('initial-isolated-default-ping',a.cmd('PING'),'PONG')
 info=a.cmd('INFO','server');version=next(line.split(':',1)[1] for line in info.splitlines() if line.startswith('redis_version:'))
 check('string-counter-increments',a.cmd('INCR','cache:counter'),1);check('string-counter-second-increment',a.cmd('INCR','cache:counter'),2)
 check('hash-add-two-fields',a.cmd('HSET','cache:profile','name','Sample','plan','free'),2)
 check('hash-update-existing-field',a.cmd('HSET','cache:profile','plan','trial'),0);check('hash-other-field-preserved',a.cmd('HGET','cache:profile','name'),'Sample')
 check('set-deduplicates',a.cmd('SADD','cache:members','one','one','two'),2);check('set-membership',a.cmd('SISMEMBER','cache:members','two'),1)
 check('sorted-set-score-order',a.cmd('ZADD','cache:rank','10','a','20','b'),2);check('sorted-set-reversed-ranking',a.cmd('ZREVRANGE','cache:rank',0,-1),['b','a'])
 check('ttl-created-with-set-ex',a.cmd('SET','cache:expires','old','EX',30),'OK');ttl('ttl-in-positive-window',a,'cache:expires')
 check('plain-set-replaces-value',a.cmd('SET','cache:expires','new'),'OK');check('plain-set-removes-existing-ttl',a.cmd('TTL','cache:expires'),-1)
 a.cmd('SET','cache:expires','old','EX',30);check('set-keepttl-update',a.cmd('SET','cache:expires','new','KEEPTTL'),'OK');ttl('keepttl-retains-positive-window',a,'cache:expires')
 a.cmd('EXPIRE','cache:profile',30);a.cmd('HSET','cache:profile','plan','paid');ttl('hash-field-write-retains-key-ttl',a,'cache:profile')
 check('missing-key-ttl',a.cmd('TTL','cache:missing'),-2)
 # Two independent connections: WATCH is invalidated by an intervening write.
 b=client();assert b.cmd('AUTH','default',bootstrap)=='OK';a.cmd('SET','cache:watch','0');check('watch-registers',a.cmd('WATCH','cache:watch'),'OK');a.cmd('MULTI');a.cmd('INCR','cache:watch');b.cmd('INCR','cache:watch');check('watch-exec-aborts-after-intervening-write',a.cmd('EXEC'),None);check('aborted-transaction-did-not-write',a.cmd('GET','cache:watch'),'1')
 a.cmd('MULTI');a.cmd('INCR','cache:watch');a.cmd('GET','cache:watch');check('multi-exec-success',a.cmd('EXEC'),[2,'2'])
 # Cache-reader example from the pinned source, using generated in-memory passwords.
 admin_password=secrets.token_hex(24);reader_password=secrets.token_hex(24)
 a.cmd('ACL','SETUSER','fixture_admin','reset','on','>'+admin_password,'~*','&*','+@all')
 a.cmd('ACL','SETUSER','cache_reader','reset','on','>'+reader_password,'~cache:*','+get','+mget','+scan')
 a.cmd('SET','cache:visible','synthetic-cache');a.cmd('SET','private:probe','synthetic-private')
 admin=client();check('dedicated-admin-authentication',admin.cmd('AUTH','fixture_admin',admin_password),'OK');admin.cmd('ACL','SETUSER','default','off')
 anon=client();check('anonymous-command-after-default-disabled',anon.cmd('GET','cache:visible'),{'error':'NOAUTH'})
 reader=client();check('reader-authentication',reader.cmd('AUTH','cache_reader',reader_password),'OK')
 check('reader-get-allowed-prefix',reader.cmd('GET','cache:visible'),'synthetic-cache');check('reader-get-denied-other-prefix',reader.cmd('GET','private:probe'),{'error':'NOPERM'})
 check('reader-write-denied',reader.cmd('SET','cache:visible','attempt'),{'error':'NOPERM'});check('reader-flushall-denied',reader.cmd('FLUSHALL'),{'error':'NOPERM'})
 check('reader-mixed-prefix-mget-denied',reader.cmd('MGET','cache:visible','private:probe'),{'error':'NOPERM'})
 check('denied-write-left-value-unchanged',admin.cmd('GET','cache:visible'),'synthetic-cache')
 names=[];cursor='0'
 for _ in range(50):
  cursor,chunk=reader.cmd('SCAN',cursor,'COUNT',100);names.extend(chunk)
  if cursor=='0':break
 assert cursor=='0'
 check('scan-can-reveal-key-name-outside-acl-prefix','private:probe' in names,True)
 # ACL SETUSER updates add rules unless the relevant permissions are reset.
 admin.cmd('ACL','SETUSER','cache_reader','~private:*','+set')
 admin.cmd('ACL','SETUSER','cache_reader','~cache:*','+get')
 check('additive-acl-update-retains-private-prefix',reader.cmd('GET','private:probe'),'synthetic-private')
 check('additive-acl-update-retains-write-command',reader.cmd('SET','cache:visible','changed'),'OK')
 check('acl-tighten-key-and-command-rules',admin.cmd('ACL','SETUSER','cache_reader','resetkeys','-@all','~cache:*','+get','+mget'),'OK')
 check('resetkeys-removes-private-prefix',reader.cmd('GET','private:probe'),{'error':'NOPERM'});check('command-reset-removes-write',reader.cmd('SET','cache:visible','attempt'),{'error':'NOPERM'})
 check('command-reset-removes-scan',reader.cmd('SCAN',0),{'error':'NOPERM'})
 admin.cmd('ACL','SETUSER','cache_reader','off')
 fresh=client();check('off-prevents-new-user-authentication',fresh.cmd('AUTH','cache_reader',reader_password),{'error':'WRONGPASS'})
 check('off-existing-authenticated-connection-still-reads',reader.cmd('GET','cache:visible'),'changed')
 check('kill-user-closes-one-existing-connection',admin.cmd('CLIENT','KILL','USER','cache_reader'),1)
 try:reader.cmd('GET','cache:visible');closed=False
 except (EOFError,ConnectionError,OSError):closed=True
 check('killed-user-connection-closed',closed,True)
 check('private-data-remains-after-denied-destructive-command',admin.cmd('GET','private:probe'),'synthetic-private')
 evidence={'format':1,'executed_at':dt.datetime.now(dt.UTC).isoformat(),'redis_version':version,'python':sys.version.split()[0],'observations':len(cases),'cases':cases,'runtime_tested':False,'native_command_fixture_tested':True,'tls_tested':False,'production_tested':False,'ai_client_executed':False,'synthetic_credentials_retained':False,'source_hashes':{'probe-redis.py':hashlib.sha256(Path(__file__).read_bytes()).hexdigest()}}
 print(json.dumps(evidence))
finally:
 for c in clients:
  try:c.close()
  except OSError:pass