Is mem0ai/mem0 safe?
- Python shell/command execution
- Node.js shell/command execution
- npm install-time lifecycle hook
What to do: Nothing here argues against installing it. Grant the capabilities it lists only if you expect the tool to need them.
mem0ai/mem0 is a PyPI package analyzed by SkillTotal's deterministic static scanner. The scan found no malicious indicators. It can: filesystem read, filesystem write, install time execution, mcp tools detected, network egress, scoped identity and shell execution — capabilities are what the code can do, not a verdict on intent. Risk score 0/100 (low).
mem0ai 2.2.1
Automated static-analysis result. It can contain false positives and false negatives, and is not a claim about the intent of mem0ai/mem0's authors. Report a false positive.
Behavioral traits
How this component maps to the CSA agentic threat model. Descriptive — it never affects the risk score.
Findings (13)
package.json runs scripts automatically when the package is installed.
"install": {Why it matters: Install scripts are a favorite supply-chain foothold — they execute on every machine that installs the package.
Fix: Inspect the hook command. Install-time scripts are a common supply chain execution vector; ensure they do nothing beyond a documented build step.
An MCP server entry launches a command on your host.
"command": "python3",
"command": "python3",
"command": "python3",
"command": "python3",
"command": "python3",
"command": "python3",
Why it matters: Trusting the manifest means running that binary — verify what it is and where it comes from.
Fix: Verify the launched command and its source before trusting this MCP server configuration.
The component can run operating-system commands or spawn processes.
import { spawn } from "node:child_process";const child = spawn(process.execPath, [SENDER_SCRIPT], {import { execFileSync } from "node:child_process";Why it matters: Powerful and often legitimate — confirm the commands aren't built from untrusted input.
Fix: Confirm the command and its arguments are fully controlled and not derived from untrusted input; prefer execFile with an argument array.
The component can run operating-system commands or spawn processes.
child = subprocess.Popen(
[sys.executable, "-m", "mem0_cli.telemetry_sender"],
stdin=subprocess.PIPE,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True, …result = subprocess.run(
command,
cwd=cwd,
env=environment,
text=True,
capture_output=True,
check=False,
)subprocess.Popen(
[sys.executable, str(worker), str(running_path)],
stdin=subprocess.DEVNULL,
stdout=log_handle, stderr=log_handle,
close_fds=True,
env=child_env,
**det …result = subprocess.run(
["git", "-C", cwd, *args],
check=False,
capture_output=True,
text=True,
timeout=0.5,
)subprocess.Popen(
[sys.executable, str(Path(__file__).resolve())],
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=True,
**mem …subprocess.Popen(
[sys.executable, str(worker), str(running_path)],
stdin=subprocess.DEVNULL,
stdout=log_handle, stderr=log_handle,
close_fds=True,
env=child_env,
**det …result = subprocess.run(
["git", "-C", cwd, *args],
check=False,
capture_output=True,
text=True,
timeout=0.5,
)subprocess.Popen(
[sys.executable, str(Path(__file__).resolve())],
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=True,
**mem …subprocess.Popen(
[sys.executable, str(worker), str(running_path)],
stdin=subprocess.DEVNULL,
stdout=log_handle, stderr=log_handle,
close_fds=True,
env=child_env,
**det …result = subprocess.run(
["git", "-C", cwd, *args],
check=False,
capture_output=True,
text=True,
timeout=0.5,
)subprocess.Popen(
[sys.executable, str(Path(__file__).resolve())],
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=True,
**mem …subprocess.Popen(
[sys.executable, str(worker), str(running_path)],
stdin=subprocess.DEVNULL,
stdout=log_handle, stderr=log_handle,
close_fds=True,
env=child_env,
**det …result = subprocess.run(
["git", "-C", cwd, *args],
check=False,
capture_output=True,
text=True,
timeout=0.5,
)subprocess.Popen(
[sys.executable, str(Path(__file__).resolve())],
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=True,
**mem …subprocess.Popen(
[sys.executable, str(worker), str(running_path)],
stdin=subprocess.DEVNULL,
stdout=log_handle, stderr=log_handle,
close_fds=True,
env=child_env,
**det …result = subprocess.run(
["git", "-C", cwd, *args],
check=False,
capture_output=True,
text=True,
timeout=0.5,
)subprocess.Popen(
[sys.executable, str(Path(__file__).resolve())],
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=True,
**mem …return subprocess.run(["docker", *args], capture_output=True, timeout=timeout, stdin=subprocess.DEVNULL, **kwargs)
subprocess.Popen([ollama_bin, "serve"], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
subprocess.run([ollama_bin or "ollama", "pull", model], timeout=600, stdin=subprocess.DEVNULL)
subprocess.Popen(
[sys.executable, str(worker), str(running_path)],
stdin=subprocess.DEVNULL,
stdout=log_handle, stderr=log_handle,
close_fds=True,
env=child_env,
**det …result = subprocess.run(
["git", "-C", cwd, *args],
check=False,
capture_output=True,
text=True,
timeout=0.5,
)subprocess.Popen(
[sys.executable, str(Path(__file__).resolve())],
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=True,
**mem …result = subprocess.run(
["git", "-C", cwd, *args],
check=False,
capture_output=True,
text=True,
timeout=0.5,
)subprocess.Popen(
[sys.executable, str(Path(__file__).resolve())],
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=True,
**mem …Why it matters: Powerful and often legitimate — confirm the commands aren't built from untrusted input.
Fix: Confirm the command and its arguments are fully controlled and not derived from untrusted input; avoid shell=True.
The component reads files from disk.
return JSON.parse(fs.readFileSync(file, 'utf8'));
const raw = fs.readFileSync(opts.file, "utf-8");
content = fs.readFileSync(0, "utf-8").trim();
const raw = fs.readFileSync(filePath, "utf-8");
const raw = fs.readFileSync(CONFIG_FILE, "utf-8");
resolvedQuery = fs.readFileSync(0, "utf-8").trim();
resolvedText = fs.readFileSync(0, "utf-8").trim();
const spec = JSON.parse(fs.readFileSync(specPath, "utf-8"));
raw = fs.readFileSync(filePath, "utf-8");
text = fs.readFileSync(filePath, "utf-8");
const raw = fs.readFileSync(configPath, "utf-8");
return JSON.parse(fs.readFileSync(new URL("../package.json", import.meta.url), "utf-8")).version;const stored = JSON.parse(fs.readFileSync(identityPath(), "utf-8"));
const stored = JSON.parse(fs.readFileSync(identityPath(), "utf-8")).anonymousId;
return fs.readFileSync(filePath, "utf-8");
fs.readFileSync(new URL("../package.json", import.meta.url), "utf-8"),const raw = fs.readFileSync(CONFIG_PATH, "utf-8");
return JSON.parse(fs.readFileSync(new URL("../package.json", import.meta.url), "utf-8")).version;const stored = JSON.parse(fs.readFileSync(TELEMETRY_ID_PATH, "utf-8")).anonymousId;
const stored = JSON.parse(fs.readFileSync(TELEMETRY_ID_PATH, "utf-8")).anonymousId;
const parsed = JSON.parse(node.fs.readFileSync(node.configPath, "utf8"));
const parsed = JSON.parse(fs.readFileSync(configPath, "utf8"));
Why it matters: Usually legitimate, but worth confirming it can't be steered into reading sensitive files.
Fix: Confirm which files are read and that paths cannot be influenced by untrusted input to reach sensitive locations.
The component writes or deletes files on disk.
fs.writeFileSync(CONFIG_FILE, JSON.stringify(data, null, 2));
fs.writeFileSync(tmp, content, "utf-8");
fs.unlinkSync(tmp);
fs.writeFileSync(configPath, JSON.stringify(cfg, null, 2));
fs.writeFileSync(identityPath(), JSON.stringify({ anonymousId: created }), "utf-8");fs.unlinkSync(identityPath());
fs.writeFileSync(filePath, content, opts);
fs.unlinkSync(filePath);
fs.writeFileSync(TELEMETRY_ID_PATH, JSON.stringify({ anonymousId: created }), "utf-8");fs.unlinkSync(TELEMETRY_ID_PATH);
node.fs.writeFileSync(node.configPath, JSON.stringify(config, null, 4));
fs.writeFileSync(tempPath, JSON.stringify(config, null, 4));
if (fs.existsSync(tempPath)) fs.unlinkSync(tempPath);
fs.rmSync(config.path, { recursive: true });Why it matters: Usually legitimate, but worth confirming the paths can't be controlled by untrusted input.
Fix: Confirm which files are written/deleted and that paths cannot be influenced by untrusted input.
The component reads files from disk.
raw = Path(file).read_text()
data = json.loads(Path(file_path).read_text())
with open(CONFIG_FILE) as f:
text = path.read_text(encoding="utf-8")
with open(config_path) as f:
rendered = render_template(source.read_text(encoding="utf-8"), values)
descriptor = json.loads(descriptor_path.read_text(encoding="utf-8"))
if not installed.is_file() or source.read_bytes() != installed.read_bytes():
value = json.loads(path.read_text(encoding="utf-8"))
schema = json.loads((SCHEMAS / schema_name).read_text(encoding="utf-8"))
json.loads(path.read_text(encoding="utf-8"))
if MONOREPO_IMPORT.search(artifact.read_text(encoding="utf-8")):
payload = json.loads(handoff_path.read_text(encoding="utf-8"))
data = json.loads((Path.home() / ".mem0" / "project_map.json").read_text())
return (data_dir() / "api-key").read_text(encoding="utf-8").strip()
_salt_cache = path.read_text(encoding="utf-8").strip()
stream.write(temporary.read_text(encoding="utf-8"))
_salt_cache = path.read_text(encoding="utf-8").strip()
value = json.loads(_identity_path().read_text(encoding="utf-8"))
state = json.loads(path.read_text(encoding="utf-8"))
lines = claim.read_text(encoding="utf-8").splitlines()
payload = json.loads(handoff_path.read_text(encoding="utf-8"))
data = json.loads((Path.home() / ".mem0" / "project_map.json").read_text())
return (data_dir() / "api-key").read_text(encoding="utf-8").strip()
_salt_cache = path.read_text(encoding="utf-8").strip()
Why it matters: Usually legitimate, but worth confirming it can't be steered into reading sensitive files.
Fix: Confirm which files are read and that paths cannot be influenced by untrusted input to reach sensitive locations.
The component writes or deletes files on disk.
with open(CONFIG_FILE, "w") as f:
_atomic_write_text(path, json.dumps(data, indent=2, ensure_ascii=False) + "\n")
_atomic_write_text(path, new_text)
os.unlink(tmp_path)
with open(config_path, "w") as f:
shutil.copytree(staged, temporary, dirs_exist_ok=True)
shutil.rmtree(output)
shutil.rmtree(temporary)
shutil.copy2(source, core / source.name)
(core / "_harness_id.py").write_text(_render_harness_id(host, portable=portable), encoding="utf-8")
target.write_text(rendered, encoding="utf-8")
shutil.copy2(source, target)
args.report.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
args.report.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
temporary.write_text(json.dumps(payload), encoding="utf-8")
log_handle = open(log_path, "a", encoding="utf-8")
temporary_path.write_text(
json.dumps({
"hook_input": hook_input,
"reason": reason,
"wait_for_inflight": wait_for_inflight,
}),
encoding="utf-8",
)temporary_path.write_text(
json.dumps({
"hook_input": hook_input,
"reason": "idle",
"delay_seconds": delay,
}),
encoding="utf-8",
)temporary.write_text(json.dumps(identity), encoding="utf-8")
temporary.write_text(
json.dumps({"plugin_version": memory_core.PLUGIN_VERSION, "repaired_at": memory_core.utc_now()}),
encoding="utf-8",
)temporary.write_text(json.dumps(state), encoding="utf-8")
with open(temporary, "w", encoding="utf-8") as handle:
temporary.write_text(json.dumps(payload), encoding="utf-8")
log_handle = open(log_path, "a", encoding="utf-8")
temporary_path.write_text(
json.dumps({
"hook_input": hook_input,
"reason": reason,
"wait_for_inflight": wait_for_inflight,
}),
encoding="utf-8",
)Why it matters: Usually legitimate, but worth confirming the paths can't be controlled by untrusted input.
Fix: Confirm which files are written/deleted and that paths cannot be influenced by untrusted input.
package.json has a 'prepare' script (runs on git/local installs and before publishing).
"prepare": "npm run build"
Why it matters: Usually a build step, but confirm it doesn't fetch or run remote code.
Fix: Usually a legitimate build step; confirm it only builds and does not fetch or execute remote code.
The component makes outbound network requests.
const resp = await fetch(url, fetchOpts);
resp = await fetch(`${baseUrl}/api/v1/auth/agent_mode/`, {const sendResp = await fetch(`${baseUrl}/api/v1/auth/email_code/`, {const verifyResp = await fetch(`${baseUrl}/api/v1/auth/email_code/verify/`, {const resp = await fetch(`${baseUrl}${path}`, {resp = await fetch(`${baseUrl}/api/v1/auth/agent_mode/caller/`, {const resp = await fetch(`${baseUrl.replace(/\/+$/, "")}/v1/ping/`, {const resp = await fetch(
const resp = await fetch(`${url}/api/v1/auth/email_code/`, {const verifyResp = await fetch(`${url}/api/v1/auth/email_code/verify/`, {const req = https.request(opts, (res) => {const response = await fetch(POSTHOG_BATCH_URL, {const resp = await fetch(url, fetchOpts);
resp = await fetch(url, {const resp = await fetch(`${baseUrl}/v1/ping/`, {const resp = await fetch(`${url.replace(/\/+$/, "")}/api/tags`, { signal: AbortSignal.timeout(3000) });const resp = await fetch(`${url.replace(/\/+$/, "")}/healthz`, { signal: AbortSignal.timeout(3000) });fetch(`${getBaseUrl().replace(/\/+$/, "")}/v1/ping/`, {const response = await fetch(`${baseUrl}/v3/memories/search/`, options);const response = await fetch(`${baseUrl}/v3/memories/add/`, options);import axios from "axios";
this.client = axios.create({const response = await fetch(url, {// fetch() reuses the pooled connection; axios here defaulted to
const response = await fetch(this.host, {Why it matters: Usually legitimate, but confirm the destinations are expected and no sensitive data leaves.
Fix: Confirm the destination hosts are expected and that no sensitive data is sent off-host.
The component makes outbound network requests.
import httpx
self._client = httpx.Client(
base_url=self.base_url,
headers={
"Authorization": f"Token {config.api_key}",
"Content-Type": "application/json",
"X-Mem0-Source": "CLI", …with httpx.Client(timeout=30.0) as client:
with httpx.Client(timeout=30.0) as client:
with httpx.Client(timeout=30.0) as client:
with httpx.Client(timeout=30.0) as client:
import httpx
resp = httpx.get(
f"{base_url.rstrip('/')}/v1/ping/",
headers={"Authorization": f"Token {api_key}"},
timeout=timeout,
)with httpx.Client(timeout=30.0) as client:
resp = httpx.patch(
f"{base_url.rstrip('/')}/api/v1/auth/agent_mode/caller/",
headers={
"Authorization": f"Token {key}",
"Content-Type": "application/js …import urllib.request
req = urllib.request.Request(
ping_url,
headers={
"Authorization": "Token " + ctx["mem0_api_key"],
"Content-Type": "application/json",
},
)resp = urllib.request.urlopen(req, timeout=10)
req = urllib.request.Request(
posthog_host,
data=body,
headers={"Content-Type": "application/json"},
)urllib.request.urlopen(req, timeout=10)
import urllib.request
parsed = urllib.parse.urlsplit(remote)
remote = urllib.parse.urlunsplit(
(parsed.scheme, hostname, parsed.path, parsed.query, parsed.fragment)
)request = urllib.request.Request(
url,
data=raw,
headers=platform_headers(key),
method="POST",
)with urllib.request.urlopen(request, timeout=timeout) as response:
request = urllib.request.Request(
url,
headers=platform_headers(key),
method="GET",
)with urllib.request.urlopen(request, timeout=timeout) as response:
Why it matters: Usually legitimate, but confirm the destinations are expected and no sensitive data leaves.
Fix: Confirm the destination hosts are expected and that no sensitive data is sent off-host.
A short-lived, scoped, assumed identity was detected — an STS AssumeRole / session token, a cloud managed or workload identity, an impersonated service account, a projected Kubernetes service-account token, or a dynamic-secret broker. Tools authenticate with a narrowly-scoped credential that expires, rather than a long-lived embedded service credential. (25 occurrence(s) shown as evidence).
* API key for authentication (if not provided, uses DefaultAzureCredential)
// Determine authentication: API key or DefaultAzureCredential
credential = new identitySdk.DefaultAzureCredential();
const { DefaultAzureCredential } = await import("@azure/identity");const credential = new DefaultAzureCredential();
description="Use Azure DefaultAzureCredential for authentication instead of password"
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
# If the API key is not provided or is a placeholder, use DefaultAzureCredential.
self.credential = DefaultAzureCredential()
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
# If the API key is not provided or is a placeholder, use DefaultAzureCredential.
self.credential = DefaultAzureCredential()
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
# If the API key is not provided or is a placeholder, use DefaultAzureCredential.
self.credential = DefaultAzureCredential()
from azure.identity import DefaultAzureCredential
# If the API key is not provided or is a placeholder, use DefaultAzureCredential.
credential = DefaultAzureCredential()
# If the API key is not provided or is a placeholder, use DefaultAzureCredential.
credential = DefaultAzureCredential()
from azure.identity import DefaultAzureCredential
use_azure_credential (bool): Use Azure DefaultAzureCredential for authentication
"""Setup Azure authentication using DefaultAzureCredential."""
credential = DefaultAzureCredential()
logger.info("Successfully authenticated using Azure DefaultAzureCredential")Fix: A scoped, short-lived identity is the smallest-blast-radius execution context. Confirm the assumed role / requested scope grants only the permissions the tool needs, and that the token lifetime is minimal.
An MCP tool surface (manifest or tool definitions) was found.
"mcpServers": {"mcpServers": {"mcpServers": {"mcpServers": {"mcpServers": {"mcpServers": {api.registerTool(createMemorySearchTool(deps), nonOptional);
api.registerTool(createMemoryAddTool(deps), nonOptional);
api.registerTool(createMemoryGetTool(deps), nonOptional);
api.registerTool(createMemoryListTool(deps), nonOptional);
api.registerTool(createMemoryUpdateTool(deps), nonOptional);
api.registerTool(createMemoryDeleteTool(deps), nonOptional);
api.registerTool(createMemoryEventListTool(deps), nonOptional);
api.registerTool(createMemoryEventStatusTool(deps), nonOptional);
pi.registerTool({Why it matters: Just context — review which tools it offers and their permissions.
Fix: Review the declared MCP tools and their permissions.
How attackers abuse these capabilities
Interactive labs on the attack class behind the rules above. They show the technique, not anything found in mem0ai/mem0.
Check your own component
Run the same evidence-backed scan on any MCP server, agent skill, or package.
Scan your own componentHow we determine this: deterministic static analysis (regex + AST), evidence-anchored, no code execution. Methodology →