class Crowbar::Session

Overview

Persistent Session Manager for iterative fuzzing and evolutionary feedback. Stores baseline input, bandit weights, genetic corpus, and mutation histories across CLI invocations.

Included Modules

Defined in:

crowbar/session.cr

Constructors

Class Method Summary

Instance Method Summary

Constructor Detail

def self.load_or_create(id : String, baseline : Buffer, rule_name : String | Nil = nil, pattern_name : String | Nil = nil, selected_mutations : String | Nil = nil, seed : UInt64 | Nil = nil, dir : String = default_dir) : Session #

def self.new(id : String, baseline : Buffer, rule_name : String | Nil = nil, pattern_name : String | Nil = nil, selected_mutations : String | Nil = nil, seed : UInt64 | Nil = nil) #

def self.new(pull : JSON::PullParser) #

def self.new(*, __pull_for_json_serializable pull : JSON::PullParser) #

Class Method Detail

def self.default_dir : String #

def self.exists?(id : String, dir : String = default_dir) : Bool #

def self.items_dir(id : String, dir : String = default_dir) : String #

def self.list(dir : String = default_dir) : Array(String) #

def self.load(id : String, dir : String = default_dir) : Session | Nil #

def self.reset(id : String, dir : String = default_dir) : Bool #

def self.session_path(id : String, dir : String = default_dir) : String #

Instance Method Detail

def active_rules : Array(String) #

def active_rules=(active_rules : Array(String)) #

def add_mutator(name : String) : self #

def add_rule(name : String) : self #

def add_scope(name : String, selector_type : String, params : Hash(String, String) = Hash(String, String).new, weight : Float64 = 1.0) : self #

def arms : Hash(String, ArmState) #

def arms=(arms : Hash(String, ArmState)) #

def auto_detect_rule : String | Nil #

def baseline : Buffer #

def baseline=(buf : Buffer) #

def baseline_b64 : String #

def baseline_b64=(baseline_b64 : String) #

def build_engine : Engine #

Builds and re-hydrates an Engine instance with stored session state


def clear_rules : self #

def clear_scopes : self #

def clear_seen_hashes : self #

def corpus_items : Array(CandidateState) #

def corpus_items=(corpus_items : Array(CandidateState)) #

def created_at : Time #

def created_at=(created_at : Time) #

def get_item(iter : Int32) : SessionItem | Nil #

Retrieves a SessionItem by iteration number


def get_mutant(iter : Int32, dir : String = Session.default_dir) : Buffer | Nil #

Retrieves the mutant buffer for an iteration (from disk cache or deterministic replay)


def history : Array(HistoryEntry) #

def history=(history : Array(HistoryEntry)) #

def id : String #

def id=(id : String) #

def input_encoding : String | Nil #

def input_encoding=(input_encoding : String | Nil) #

def item_retention : Int32 #

def item_retention=(item_retention : Int32) #

def items : Array(SessionItem) #

def items=(items : Array(SessionItem)) #

def items_dir(dir : String = Session.default_dir) : String #

def iteration : Int32 #

def iteration=(iteration : Int32) #

def last_mutant : Buffer | Nil #

def last_mutant=(buf : Buffer | Nil) #

def last_mutant_b64 : String | Nil #

def last_mutant_b64=(last_mutant_b64 : String | Nil) #

def last_mutators : Array(String) #

def last_mutators=(last_mutators : Array(String)) #

def last_reward : Float64 | Nil #

def last_reward=(last_reward : Float64 | Nil) #

def next_mutant(dir : String = Session.default_dir, unique : Bool | Nil = nil, seek : Int64 | Nil = nil, template_override : String | Nil = nil) : Buffer #

Generates the next mutant, updates session iteration, syncs bandit & corpus, and saves state. Optionally accepts runtime overrides for uniqueness deduplication, PRNG seek, and output templating.


def output_encoding : String | Nil #

def output_encoding=(output_encoding : String | Nil) #

def pattern_name : String | Nil #

def pattern_name=(pattern_name : String | Nil) #

def remove_mutator(name : String) : self #

def remove_rule(name : String) : self #

def remove_scope(name : String) : self #

def replay(iter : Int32) : Tuple(Buffer, Array(TransformationStep)) #

Deterministically replays and reconstructs an iteration from baseline


def reset_mutators : self #

def reset_with(new_baseline : Buffer, rule_name : String | Nil = nil, pattern_name : String | Nil = nil, selected_mutations : String | Nil = nil, seed : UInt64 | Nil = nil) #

Resets the session with a new baseline input, clearing corpus and iteration counter


def reward(value : Float64, dir : String) : Session #

Applies feedback score (-1.0 to 1.0) with custom directory override


def reward(value : Float64, iteration : Int32 | Nil = nil, dir : String = Session.default_dir) : Session #

Applies feedback score (-1.0 to 1.0) to a specific generated mutant (default: latest)


def rule_name : String | Nil #

def rule_name=(rule_name : String | Nil) #

def save(dir : String = Session.default_dir) #

Serializes session to JSON file on disk


def scopes : Array(ScopeConfig) #

def scopes=(scopes : Array(ScopeConfig)) #

def seed : UInt64 #

def seed=(seed : UInt64) #

def seek_offset : Int64 #

def seek_offset=(seek_offset : Int64) #

def seen_hashes : Array(UInt64) #

def seen_hashes=(seen_hashes : Array(UInt64)) #

def selected_mutations : String | Nil #

def selected_mutations=(selected_mutations : String | Nil) #

def set_mutators(list : Array(String)) : self #

def set_pattern(pat : String) : self #

def set_seek(offset : Int64) : self #

def set_template(spec : String | Nil) : self #

def set_unique(enabled : Bool, capacity : Int32 = 10000) : self #

def template : String | Nil #

def template=(template : String | Nil) #

def total_pulls : Int64 #

def total_pulls=(total_pulls : Int64) #

def unique_enabled : Bool #

def unique_enabled=(unique_enabled : Bool) #

def uniqueness_capacity : Int32 #

def uniqueness_capacity=(uniqueness_capacity : Int32) #

def updated_at : Time #

def updated_at=(updated_at : Time) #