Repository navigation
Add ScorpioSync optional addon for cover traffic generation and traffic-level camouflage #7
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from all commits
Commits
Show all changes
3 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,3 @@ | ||
| """ | ||
| Addons package for n'éccrit - Optional extensions for traffic camouflage. | ||
| """ |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| """ | ||
| Adapter for integrating scorpiosync generators with n'éccrit transport layer. | ||
|
|
||
| This module provides a bridge between the cover traffic generators and the | ||
| existing spectralship_engine_duallayer transport mechanisms. | ||
| """ | ||
| import json, random, threading, time | ||
| from typing import Optional | ||
| from .scorpiosync_generators_complete import fake_api_request_generator, http_builder_transformer, steganography_transformer | ||
|
|
||
|
|
||
| # Global state for session tracking (to be set by the application) | ||
| _active_node = None | ||
| _peer_config = None | ||
|
|
||
|
|
||
| def set_active_node(node, peer_ip: str, peer_port: int, peer_name: str): | ||
| """ | ||
| Configure the active node and peer for cover traffic transmission. | ||
|
|
||
| Args: | ||
| node: SpectralNode instance | ||
| peer_ip: Peer IP address | ||
| peer_port: Peer port | ||
| peer_name: Peer name | ||
| """ | ||
| global _active_node, _peer_config | ||
| _active_node = node | ||
| _peer_config = {"ip": peer_ip, "port": peer_port, "name": peer_name} | ||
|
|
||
|
|
||
| def is_session_ready() -> bool: | ||
| """Check if a session is ready for cover traffic transmission.""" | ||
| return _active_node is not None and _peer_config is not None and _active_node.running | ||
|
|
||
|
|
||
| def send_encrypted_json(frame: dict): | ||
| """ | ||
| Send a cover traffic frame via the active session. | ||
|
|
||
| The frame is sent as a regular spectralship message with type='cover'. | ||
| This maintains compatibility with the existing transport layer. | ||
| """ | ||
| if not is_session_ready(): | ||
| raise RuntimeError("No active session available for cover traffic") | ||
|
|
||
| # Convert the cover frame to a message that can be sent via spectralship | ||
| # We use a minimal signal to encode just the type information | ||
| cover_msg = f"[COVER:{frame.get('type', 'cover')}]" | ||
| _active_node.send_fragment( | ||
| cover_msg, | ||
| _peer_config["ip"], | ||
| _peer_config["port"], | ||
| _peer_config["name"] | ||
| ) | ||
|
|
||
|
|
||
| def maybe_transform_plaintext(plaintext: str, use_stegano: bool) -> str: | ||
| """ | ||
| Optionally apply steganographic transformation to plaintext. | ||
|
|
||
| Args: | ||
| plaintext: Input text | ||
| use_stegano: Whether to apply zero-width steganography | ||
|
|
||
| Returns: | ||
| Transformed or original plaintext | ||
| """ | ||
| return steganography_transformer(plaintext) if use_stegano else plaintext | ||
|
|
||
|
|
||
| def send_cover_once(use_http_mimicry: bool): | ||
| """ | ||
| Send a single cover traffic packet. | ||
|
|
||
| Args: | ||
| use_http_mimicry: Whether to format as HTTP request | ||
| """ | ||
| payload = fake_api_request_generator() | ||
| frame = {"type": "cover", "payload": payload, "ts": int(time.time())} | ||
| if use_http_mimicry: | ||
| # Generate HTTP-formatted bytes (for potential future use/logging) | ||
| _ = http_builder_transformer(payload, include_length_prefix=False) | ||
| send_encrypted_json(frame) | ||
|
|
||
|
|
||
| def start_cover_traffic(min_s: int, max_s: int, use_http_mimicry: bool, stop_flag: threading.Event) -> threading.Thread: | ||
| """ | ||
| Start a background thread that sends periodic cover traffic. | ||
|
|
||
| Args: | ||
| min_s: Minimum interval between cover packets (seconds) | ||
| max_s: Maximum interval between cover packets (seconds) | ||
| use_http_mimicry: Whether to use HTTP-mimicking format | ||
| stop_flag: Threading event to signal stop | ||
|
|
||
| Returns: | ||
| Thread handle for the cover traffic generator | ||
| """ | ||
| def _loop(): | ||
| while not stop_flag.is_set(): | ||
| if is_session_ready(): | ||
| try: | ||
| send_cover_once(use_http_mimicry) | ||
| except Exception: | ||
| pass # Silently ignore errors to avoid disrupting main communication | ||
| time.sleep(random.randint(min_s, max_s)) | ||
|
|
||
| t = threading.Thread(target=_loop, daemon=True) | ||
| t.start() | ||
| return t | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| from __future__ import annotations | ||
| import json, random, secrets, string, time | ||
| from typing import Any, Dict, Optional | ||
|
|
||
| def _rand_string(length: int = 8) -> str: | ||
| return "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(length)) | ||
|
|
||
| def _now_iso() -> str: | ||
| return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) | ||
|
|
||
| def fake_api_request_generator(seed: Optional[int] = None) -> Dict[str, Any]: | ||
| if seed is not None: | ||
| random.seed(seed) | ||
| endpoints = ["/v1/users/login","/v1/items/search","/api/v2/metrics/submit"] | ||
| methods = ["POST","GET","PUT"] | ||
| endpoint = random.choice(endpoints) | ||
| method = random.choice(methods) | ||
| body = {"sample": _rand_string(6), "ts": _now_iso()} | ||
| headers = {"Host": "api.example.com", "User-Agent": "neccrit/1.0"} | ||
| return {"method": method, "endpoint": endpoint, "headers": headers, "body": body, "generated_at": _now_iso()} | ||
|
|
||
| def http_builder_transformer(payload: Any, host="api.example.com", include_length_prefix=True) -> bytes: | ||
| body = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode() | ||
| header = f"POST {payload.get('endpoint','/')} HTTP/1.1\r\nHost: {host}\r\nContent-Type: application/json\r\nContent-Length: {len(body)}\r\n\r\n".encode() | ||
| req = header + body | ||
| return len(req).to_bytes(4,"big") + req if include_length_prefix else req | ||
|
|
||
| _ZERO_WIDTH_ONE, _ZERO_WIDTH_ZERO = "\u200B", "\u200C" | ||
| def steganography_transformer(payload: Any) -> str: | ||
| data = json.dumps(payload,separators=(",",":"),ensure_ascii=False).encode() | ||
| bits = "".join(f"{b:08b}" for b in data) | ||
| return "".join(_ZERO_WIDTH_ONE if bit=="1" else _ZERO_WIDTH_ZERO for bit in bits) | ||
|
|
||
| def steganography_decode(zws: str) -> str: | ||
| bits = ["1" if c==_ZERO_WIDTH_ONE else "0" for c in zws if c in (_ZERO_WIDTH_ONE,_ZERO_WIDTH_ZERO)] | ||
| bytes_out = [int("".join(bits[i:i+8]),2) for i in range(0,len(bits),8)] | ||
| return bytes(bytes_out).decode("utf-8") | ||
|
|
||
| __all__ = ["fake_api_request_generator","http_builder_transformer","steganography_transformer","steganography_decode"] |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| """ | ||
| Configuration file for n'éccrit optional addons. | ||
|
|
||
| This file contains configuration flags for optional features like | ||
| cover traffic generation, HTTP mimicry, and steganography. | ||
|
|
||
| ⚠️ SECURITY NOTE: | ||
| - Cover traffic generation can help obfuscate communication patterns | ||
| - Zero-width steganography is DISABLED by default for responsible use | ||
| - All cover traffic is still encrypted through the standard crypto layer | ||
| """ | ||
|
|
||
| # Cover Traffic Configuration | ||
| # --------------------------- | ||
| # Enable periodic cover traffic to obfuscate communication patterns | ||
| ENABLE_COVER_TRAFFIC = True | ||
|
|
||
| # Minimum interval between cover traffic packets (seconds) | ||
| COVER_MIN_INTERVAL_S = 20 | ||
|
|
||
| # Maximum interval between cover traffic packets (seconds) | ||
| COVER_MAX_INTERVAL_S = 45 | ||
|
|
||
| # HTTP Mimicry | ||
| # ------------ | ||
| # Format cover traffic as HTTP-like requests for better camouflage | ||
| ENABLE_HTTP_MIMICRY = True | ||
|
|
||
| # Zero-Width Steganography | ||
| # ------------------------ | ||
| # ⚠️ DISABLED BY DEFAULT - Use responsibly for legitimate research only | ||
| # This feature embeds data in zero-width unicode characters | ||
| ENABLE_ZW_STEGANO = False |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The adapter converts every cover frame to the string
"[COVER:…]"and forwards it withsend_fragment. On the receiving side,SpectralNode.start_listeningtreats any incoming fragment as a normal chat message and prints/appends it, so enabling cover traffic will spam peers with visible[COVER:cover]messages instead of silently camouflaging traffic. This defeats the purpose of the addon and advertises that cover traffic is being generated. Either embed a distinguishable frame type that the receiver can ignore or add filtering on receive so cover packets are not surfaced to users.Useful? React with 👍 / 👎.