Blame for model.py

"""Authored ReLU secret scorer. No optimizer, pretrained tensors or corpus lookup."""
from pathlib import Path
import re,math,json,hashlib
from collections import Counter
import numpy as np
ROOT=Path(__file__).resolve().parent
POSITIONS=40
FLAGS=['len8','len16','len20','len24','len32','len40','len64','max128','max256','upperdigit','alnumish','mixed','entropy25','entropy3','entropy4','secret_context','password_context','bearer_context','placeholder','public_context','hash_context','uri_password','pem','jwt_shape','colon_numeric_head','colon_long_tail','semicolon_chain','long_assignment','near_secret','near_password','reference_shape','case_mix','uuid_shape']
DIM=POSITIONS*128+len(FLAGS)
F={f:POSITIONS*128+i for i,f in enumerate(FLAGS)}
# Prefix literals become first-layer coefficients on ASCII-position one-hot inputs.
PREFIXES=[('AKIA',['len20','upperdigit']),('ASIA',['len20','upperdigit']),('ABIA',['len20','upperdigit']),('ACCA',['len20','upperdigit']),('ghp_',['len32']),('gho_',['len32']),('ghu_',['len32']),('ghs_',['len32']),('ghr_',['len32']),('github_pat_',['len40']),('glpat-',['len16']),('SG.',['len40']),('xoxb-',['len20']),('xoxp-',['len20']),('xoxa-',['len20']),('xoxs-',['len20']),('sk_live_',['len24']),('rk_live_',['len24']),('sk_test_',['len24']),('sk-proj-',['len32']),('sk-ant-',['len32']),('hf_',['len24']),('npm_',['len24']),('pypi-',['len32']),('AIza',['len32']),('sq0atp-',['len20']),('dop_v1_',['len32']),('SK',['len32']),('shpat_',['len32']),('shpca_',['len32']),('shppa_',['len32']),('shpss_',['len32']),('https://hooks.slack.com/services/',['len64']),('https://hooks.slack-gov.com/services/',['len64']),('sk-',['len40']),('TlRMTVNTUA',['len64']),('LS0tLS1CRUdJTiBQUklWQVRFIEtFWS0tLS0t',['len64']),('DefaultEndpointsProtocol=',['len64','semicolon_chain','long_assignment'])]

def authored_weights():
    rows=[];bias=[];out=[];names=[]
    def gate(name,idx,weight,absent=()):
        row=np.zeros(DIM,np.float32)
        for i in idx:row[i]+=1
        for i in absent:row[i]-=1
        rows.append(row);bias.append(1-len(idx));out.append(weight);names.append(name)
    def flaggate(name,flags,weight,absent=()):
        gate(name,[F[f] for f in flags],weight,[F[f] for f in absent])
    for prefix,flags in PREFIXES:
        # Offset the plain-address inhibitor for recognized credential URL formats.
        gate('prefix:'+prefix,[i*128+ord(c) for i,c in enumerate(prefix)]+[F[f] for f in flags+['entropy3','max256']],18 if prefix.startswith(('http://','https://')) else 2)
    for name,flags,w in [('numeric_colon_credential',['colon_numeric_head','colon_long_tail','len40','entropy4','max128'],2),('contextual_random',['secret_context','len16','entropy3','mixed','max256'],2),('password',['password_context','len8','entropy25','max128'],2),('bearer',['bearer_context','len16','entropy3','max256'],2),('jwt',['jwt_shape','len32'],2),('connection',['uri_password'],2),('private_key',['pem'],2),('prose_secret',['near_secret','len16','entropy4','mixed','max256'],2),('context_uuid',['near_secret','uuid_shape'],2),('password_case',['near_password','len8','entropy4','case_mix','max128'],2),('prose_password',['near_password','len8','entropy25','mixed','max128'],2),('reference',['reference_shape'],-64),('placeholder',['placeholder'],-64)]:
        flaggate(name,flags,w)
    # Public/hash contexts inhibit ambiguous generic scores, not self-identifying formats.
    for flag in ['public_context','hash_context']:
        flaggate(flag,[flag],-4,['uri_password','pem','jwt_shape'])
    # Hexadecimal secrets in prose have at most four bits of character entropy.
    row=np.zeros(DIM,np.float32)
    for i in range(32):
        for c in '0123456789abcdefABCDEF':row[i*128+ord(c)]=1
    for f in ['near_secret','len32','mixed','max128']:row[F[f]]=1
    rows.append(row);bias.append(-35);out.append(2);names.append('prose_hex_secret')
    # Syntax inhibitors use positional observations, never benchmark credential literals.
    for i in range(POSITIONS):
        for context in ['near_password','secret_context']:
            gate('member_reference:'+context+':'+str(i),[F[context],F['case_mix'],i*128+ord('.')],-8,[F['entropy4'],F['uri_password'],F['jwt_shape']])
        if i<POSITIONS-1:
            gate('interpolated_connection:'+str(i),[F['uri_password'],i*128+ord('$'),(i+1)*128+ord('{')],-64)
    # Shell variable references exclude numeric crypt prefixes such as $2b$.
    for c in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_':
        gate('shell_reference:'+c,[F['near_password'],ord('$'),128+ord(c)],-64,[F['mixed'],F['case_mix']])
    for literal in [':password@',':pass@']:
        for start in range(POSITIONS-len(literal)+1):
            gate('placeholder_connection:'+literal+':'+str(start),[F['uri_password']]+[(start+i)*128+ord(c) for i,c in enumerate(literal)],-64)
    # Semicolon-delimited key-value credentials need not begin with one provider field.
    flaggate('connection_assignment',['semicolon_chain','long_assignment','len64','entropy4'],2)
    # A lowercase snake-case phrase after Basic/Bearer is a reference, not an encoded header.
    for i in range(POSITIONS):
        gate('bearer_reference:'+str(i),[F['bearer_context'],i*128+ord('_')],-8,[F['mixed'],F['case_mix']])
    gate('shell_interpolation',[F['near_password'],ord('$'),128+ord('{')],-64)
    gate('password_path',[F['near_password'],ord('/')],-8,[F['uri_password'],F['entropy4']])
    for prefix in ['LS0tLS1CRUdJTiBQUklWQVRFIEtFWS0tLS0t','LS0tLS1CRUdJTiBSU0EgUFJJVkFURSBLRVkt']:
        gate('encoded_private_key:'+prefix,[i*128+ord(c) for i,c in enumerate(prefix)]+[F['len64'],F['entropy4']],2)
    # Plain web addresses are context, while URLs carrying userinfo remain secrets.
    for prefix in ['http://','https://']:
        gate('plain_web_address:'+prefix,[i*128+ord(c) for i,c in enumerate(prefix)],-16,[F['uri_password'],F['jwt_shape']])
    # A rooted path needs four visible separators; a single base64 slash is insufficient.
    row=np.zeros(DIM,np.float32)
    row[ord('/')]=40
    for i in range(1,POSITIONS):row[i*128+ord('/')]=1
    for f in ['uri_password','jwt_shape']:row[F[f]]=-40
    rows.append(row);bias.append(-42);out.append(-16);names.append('rooted_source_path')
    # Scheme-relative domain references begin with // and contain a hostname dot.
    row=np.zeros(DIM,np.float32)
    row[ord('/')]=40;row[128+ord('/')]=40
    for i in range(2,POSITIONS):row[i*128+ord('.')]=1
    for f in ['uri_password','jwt_shape']:row[F[f]]=-80
    rows.append(row);bias.append(-80);out.append(-16);names.append('scheme_relative_domain')
    # Configuration filenames are references to stored credentials, not their values.
    for i in range(POSITIONS-4):
        gate('json_filename:'+str(i),[(i+j)*128+ord(c) for j,c in enumerate('.json')],-16,[F['uri_password'],F['jwt_shape']])
    # Placeholder userinfo follows the same rule regardless of uppercase spelling.
    for literal in [':PASSWORD@',':PASS@']:
        for start in range(POSITIONS-len(literal)+1):
            gate('placeholder_connection:'+literal+':'+str(start),[F['uri_password']]+[(start+i)*128+ord(c) for i,c in enumerate(literal)],-64)
    # Precision circuits: credential-adjacent source references are not values.
    # These rows require syntax structure and never encode a corpus token.
    protected=['uri_password','pem','jwt_shape']
    for context in ['secret_context','near_secret','password_context','near_password']:
        # Dotted lowercase identifiers without digits or high character entropy.
        for i in range(1,POSITIONS-1):
            gate('dotted_identifier:'+context+':'+str(i),
                 [F[context],F['alnumish'],i*128+ord('.')],-8,
                 [F[f] for f in protected+['mixed','case_mix','entropy4']])
        # Empty call/subscript delimiters are a strong code-expression cue.
        for literal in ['()','[]']:
            for start in range(POSITIONS-len(literal)+1):
                gate('code_delimiters:'+context+':'+literal+':'+str(start),
                     [F[context]]+[(start+i)*128+ord(c) for i,c in enumerate(literal)],-16,
                     [F[f] for f in protected])
        for prefix in ['self.','this.','config.','settings.','System.getenv(','getenv(']:
            gate('source_reference:'+context+':'+prefix,
                 [F[context]]+[i*128+ord(c) for i,c in enumerate(prefix)],-32,
                 [F[f] for f in protected])
    # Configuration/certificate names frequently occur alongside secret labels.
    # A URI, JWT, or PEM observation switches these filename circuits off.
    for suffix in ['.yaml','.yml','.toml','.conf','.pem','.crt','.pub']:
        for start in range(POSITIONS-len(suffix)+1):
            gate('credential_filename:'+suffix+':'+str(start),
                 [(start+i)*128+ord(c) for i,c in enumerate(suffix)],-16,
                 [F[f] for f in protected])
    # Uppercase snake-case password assignments are usually environment names.
    # Requiring an uppercase initial, underscore, and no lowercase/digits avoids
    # suppressing ordinary mixed-format password strings.
    row=np.zeros(DIM,np.float32)
    row[F['password_context']]=40
    for c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ':row[ord(c)]=40
    for i in range(1,POSITIONS):row[i*128+ord('_')]=1
    for f in protected+['mixed','case_mix']:row[F[f]]=-80
    rows.append(row);bias.append(-80);out.append(-8);names.append('uppercase_password_symbol')
    # Exploratory authored UUID-format detector, excluding explicit nonsecret contexts.
    flaggate('uuid_format_experiment',['uuid_shape','len32','entropy3','max128'],2,['public_context','hash_context'])
    return np.stack(rows),np.array(bias,np.float32),np.array(out,np.float32),names

def candidates(text):
    found={(m.start(),m.end()) for m in re.finditer(r'[A-Za-z0-9_+/.@%!?$~#*^-]{4,}={0,2}',text)}
    for pattern in [r'\b[0-9]{5,16}:[A-Za-z0-9_-]{24,80}\b',r'\b[A-Za-z][A-Za-z0-9_]*=[^\s\'\"<>;]+(?:;[A-Za-z][A-Za-z0-9_]*=[^\s\'\"<>;]+){2,};?',r'\b[a-zA-Z][a-zA-Z0-9+.-]*://[^\s\'"<>]+',r'-----BEGIN (?:[A-Z0-9 ]+ )?PRIVATE KEY-----[\s\S]*?-----END (?:[A-Z0-9 ]+ )?PRIVATE KEY-----']:
        found.update((m.start(),m.end()) for m in re.finditer(pattern,text))
    # Quoted credentials may contain separators that ordinary atom spans omit.
    for match in re.finditer(r"(['\"])([^'\"\n]{8,128})\1",text):
        if not any(c.isspace() for c in match.group(2)):found.add((match.start(2),match.end(2)))
    # Sentence punctuation and code backticks are lexical span boundaries.
    for a,b in list(found):
        if text[b-1] in '.,': found.add((a,b-1))
    for match in re.finditer(r'`([^`\n]{6,128})`',text):found.add((match.start(1),match.end(1)))
    return sorted(found)

def feature(text,a,b):
    value=text[a:b];low=value.lower();n=len(value);x=np.zeros(DIM,np.float32)
    for i,c in enumerate(value[:POSITIONS]):
        if ord(c)<128:x[i*128+ord(c)]=1
    count=Counter(value);entropy=-sum((c/n)*math.log2(c/n) for c in count.values())
    left=text[max(0,a-100):a].split('\n')[-1]
    # Only lexical observations; these do not assign a SECRET label.
    assignment=re.search(r'([A-Za-z_][A-Za-z0-9_.-]*)[\s\'"`]*[:=][\s\'"`]*$',left)
    key=assignment.group(1).lower() if assignment else ''
    placeholder=any(v in low for v in ['example','redacted','placeholder','your-api','your_api','changeme','change-me','dummy','xxxxxx','1234567890']) or len(count)<4 or low in ['undefined','password','secret','null','none'] or (a>0 and text[a-1] in '${')
    context=left[-75:]
    # General language cues; conjunction gates combine these with randomness.
    near_secret=bool(re.search(r'(?i)(?:api[ _-]?key|access[ _-]?key|secret|token|credential|api[ _-]?secret|密钥|令牌|凭证|ключ|токен|chave|jeton|clave)[^.!?\n]{0,55}$',context))
    near_password=bool(re.search(r'(?i)(?:password|passwd|pwd|mot de passe|passwort|senha|contrase[ñn]a|пароль|парол|密码|密碼|パスワード|비밀번호)[^.!?\n]{0,55}$',context))
    # Variable interpolation and code references are syntax, not literal values.
    reference_shape=bool(re.fullmatch(r'\$\{[A-Za-z_][A-Za-z0-9_]*(?::[^}]*)?\}|\{\{[A-Za-z_][A-Za-z0-9_.]*\}\}|<[^<>]+>',value)) or bool(re.match(r'(?:process\.env\.|os\.environ\.|ENV\.)[A-Za-z_]',value))
    flags={f'len{k}' :n>=k for k in [8,16,20,24,32,40,64]}
    flags.update(case_mix=any(c.isupper() for c in value) and any(c.islower() for c in value),uuid_shape=bool(re.fullmatch(r'[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}',value)),near_secret=near_secret,near_password=near_password,reference_shape=reference_shape,colon_numeric_head=bool(re.match(r'^[0-9]{5,16}:',value)),colon_long_tail=bool(re.fullmatch(r'[^:]+:[A-Za-z0-9_-]{24,80}',value)),semicolon_chain=value.count(';')>=2,long_assignment=bool(re.search(r'=[A-Za-z0-9+/]{40,}={0,2}(?:;|$)',value)),max128=n<=128,max256=n<=256,upperdigit=all(c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789' for c in value),alnumish=all(c.isalnum() or c in '_-+/.=' for c in value),mixed=any(c.isalpha() for c in value) and any(c.isdigit() for c in value),entropy25=entropy>=2.5,entropy3=entropy>=3,entropy4=entropy>=4,secret_context=any(c in key for c in ['secret','token','api_key','apikey','access_key','credential','auth']),password_context=any(c in key for c in ['password','passwd','pwd']),bearer_context=bool(re.search(r'(?i)\b(?:bearer|basic)\s+$',left)),placeholder=placeholder,public_context=any(c in key for c in ['public','publishable','client_id']),hash_context=any(c in key for c in ['sha256','sha512','checksum','commit','digest','md5']),uri_password=bool(re.match(r'[a-zA-Z][\w+.-]*://[^\s/:@]+:[^\s/@]{4,}@',value)),pem=value.startswith('-----BEGIN ') and 'PRIVATE KEY-----' in value,jwt_shape=bool(re.fullmatch(r'eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+',value)))
    for key,on in flags.items():x[F[key]]=float(on)
    return x

class Detector:
    def __init__(self,zero=False,coreml=False):
        z=np.load(ROOT/'weights.npz');self.W=z['W'];self.b=z['b'];self.o=z['o']*(0 if zero else 1)
        self.nz=[np.flatnonzero(row) for row in self.W];self.native=None
        if coreml:
            import coremltools as ct
            self.native=ct.models.MLModel(str(ROOT/'bundle/BlotHandwritten.mlpackage'),compute_units=ct.ComputeUnit.CPU_ONLY)
    def scores(self,X):
        if self.native:
            values=[]
            for start in range(0,len(X),128):
                chunk=np.zeros((128,DIM),np.float32);v=X[start:start+128];chunk[:len(v)]=v
                values.extend(self.native.predict({'features':chunk})['scores'][:len(v),0].tolist())
            return np.array(values)
        # Exact sparse evaluation of the same frozen matrices; no alternate rules.
        hidden=np.stack([X[:,idx]@self.W[i,idx]+self.b[i] for i,idx in enumerate(self.nz)],axis=1)
        return np.maximum(hidden,0)@self.o-0.5
    def predict(self,text):
        spans=candidates(text)
        if not spans:return []
        scores=self.scores(np.stack([feature(text,a,b) for a,b in spans]))
        selected=[]
        for (a,b),score in sorted(zip(spans,scores),key=lambda v:(-v[1],-(v[0][1]-v[0][0]),v[0][0])):
            if score>0 and not any(a<y and b>x for x,y,_ in selected):selected.append((a,b,'SECRET'))
        return sorted(selected)

if __name__=='__main__':
    import sys
    detector=Detector(coreml='--coreml' in sys.argv)
    for line in sys.stdin:
        text=json.loads(line)['text'];print(json.dumps([dict(start=a,end=b,label=c) for a,b,c in detector.predict(text)]))