~/.kognisant_core/scripts/legit-bot.py → /etc/cron.d/malicious

def _resolve_script_path(script_path: str) -> str:
    """Resolve and validate script path against symlink attacks.

    1. Construct full path: SCRIPTS_DIR / script_path
    2. Resolve ALL symlinks: os.path.realpath(full_path)
    3. Verify resolved path starts with SCRIPTS_DIR

    Returns:
        Resolved absolute path (safe to execute).

    Raises:
        SecurityError: If resolved path escapes the scripts directory.
    """
    full_path = os.path.join(SCRIPTS_DIR, script_path)
    real_path = os.path.realpath(full_path)

    if not real_path.startswith(os.path.realpath(SCRIPTS_DIR) + os.sep):
        raise SecurityError(
            f"Script path resolves outside allowed directory: {real_path}"
        )

    return real_path

# In ProcessManager.spawn():
if not ProcessManager.check_symlink(script_path, SCRIPTS_DIR):
    job_queue.update_status(job_name, "failed",
                           error="Script path resolves outside allowed directory")
    logger.warning(f"SECURITY: Symlink escape attempt for job '{job_name}': {script_path}")
    return None

   os.chmod(tmp_path, 0o600)  # Set before file becomes visible

   os.rename(tmp_path, QUEUE_PATH)  # Now visible with correct permissions

   shutil.copy2(QUEUE_PATH, BACKUP_PATH)

   os.chmod(BACKUP_PATH, 0o600)  # Ensure backup has same protection

def _check_root_privileges():
    """Warn if running as root."""
    if os.geteuid() == 0:
        logger.warning(
            "Daemon running with root privileges. "
            "Recommend running under a non-root user account."
        )
        # Also display to terminal (before fork, in CLI)
        print("Warning: Daemon running with root privileges. "
              "Recommend running under a non-root user.", file=sys.stderr)

def _resolve_safe_path(file_path: str, project_root: str) -> str:
    """Resolve file path and verify it's within the project root.

    Protection against:
    - Relative traversal: "../../../etc/passwd"
    - Absolute paths: "/etc/passwd"
    - Symlink escape: "link_to_etc/passwd"
    """
    # Resolve both paths to eliminate symlinks
    real_root = os.path.realpath(project_root)
    real_target = os.path.realpath(os.path.join(project_root, file_path))

    # Verify target is within root
    if not real_target.startswith(real_root + os.sep) and real_target != real_root:
        return None  # Access denied

    return real_target

# 1. Create a dedicated secrets directory
mkdir -p ~/.secrets
chmod 700 ~/.secrets

2. Create the env file with restricted permissions
cat > ~/.secrets/my-bot.env << 'EOF'
Bot credentials
TELEGRAM_TOKEN=123456:ABC-DEF
DATABASE_URL=postgres://user:pass@localhost/db
API_SECRET=sk-very-secret-key
EOF
chmod 600 ~/.secrets/my-bot.env

3. Reference it when creating the job
kognisant job add --name my-bot --script bot.py --type persistent \
    --env-file ~/.secrets/my-bot.env

# Lines starting with # are comments
Blank lines are ignored
Format: KEY=VALUE (no quotes needed around value)

API_KEY=sk-abc123
DB_URL=postgres://user:pass@localhost/db
DEBUG=false

Inline comments are NOT supported:
BAD: API_KEY=sk-abc123  # this would include "  # this would include" in the value

def create_script(name: str, content: str, description: str = "",
                  env_vars: list[str] | None = None) -> str:
    """Create script atomically.

    Sequence:
    1. Write content → {name}.py.tmp
    2. Write metadata → {name}.json.tmp
    3. Rename {name}.py.tmp → {name}.py     (atomic)
    4. Rename {name}.json.tmp → {name}.json  (atomic)

    On failure at any step:
    - Remove all .tmp files
    - Leave no partial artifacts
    """

try:
    # Write temps
    write_temp_py(...)
    write_temp_json(...)
    # Rename (atomic)
    os.rename(tmp_py, final_py)
    os.rename(tmp_json, final_json)
except Exception:
    # Clean up any temps that were created
    for tmp in [tmp_py, tmp_json]:
        if os.path.exists(tmp):
            os.unlink(tmp)
    raise

import fcntl

fd = open("jobs.lock", "w")
fcntl.flock(fd, fcntl.LOCK_EX)  # Exclusive lock - blocks until acquired
... critical section ...
fcntl.flock(fd, fcntl.LOCK_UN)  # Release
fd.close()

fcntl.flock(self._fd, fcntl.LOCK_EX | fcntl.LOCK_NB)

Error: timeout - Could not acquire file lock within 5 seconds.
Another process may be holding the lock.