<b>Hello</b>
we're Invareo, a technology consultancy

Expertise in web, api, and systems development; software purchasing and implementation; cyber security, and privacy.

core/encryption.py

1	import os, hashlib, hmac
2	from cryptography.hazmat.primitives.ciphers.aead import AESGCM
3	from cryptography.hazmat.primitives.kdf.hkdf import HKDF
4	from cryptography.hazmat.primitives import hashes
5	from base64 import b64encode, b64decode
6	from typing import Optional
7
8
9	class VaultCipher:
10	"""AES-256-GCM authenticated encryption with
11	HKDF key derivation and HMAC signing."""
12
13	def __init__(self, master_key: bytes):
14	assert len(master_key) >= 32
15	self._master = master_key
16	self._enc_key = self._derive(b"encrypt")
17	self._sig_key = self._derive(b"signing")
18	self._gcm = AESGCM(self._enc_key)
19
20	def _derive(self, info: bytes) -> bytes:
21	return HKDF(
22	algorithm=hashes.SHA256(),
23	length=32,
24	salt=None,
25	info=info,
26	).derive(self._master)
27
28	def _sign(self, data: bytes) -> bytes:
29	return hmac.new(
30	self._sig_key, data, hashlib.sha256
31	).digest()
32
33	def encrypt(
34	self,
35	plaintext: bytes,
36	aad: Optional[bytes] = None,
37	) -> str:
38	nonce = os.urandom(12)
39	ct = self._gcm.encrypt(nonce, plaintext, aad)
40	blob = nonce + ct
41	sig = self._sign(blob)
42	return b64encode(sig + blob).decode()
43
44	def decrypt(
45	self,
46	ciphertext: str,
47	aad: Optional[bytes] = None,
48	) -> bytes:
49	raw = b64decode(ciphertext)
50	sig, blob = raw[:32], raw[32:]
51	if not hmac.compare_digest(
52	sig, self._sign(blob)
53	):
54	raise ValueError("HMAC verification failed")
55	return self._gcm.decrypt(
56	blob[:12], blob[12:], aad
57	)

What We Do

Our Services

Engineering Maturity & Transformation

Assess how your engineering organization actually delivers—using our comprehensive, research-backed maturity framework—and translate that into a clear, prioritized path to higher velocity and reliability. This is not a generic audit; it’s a grounded evaluation tied to your systems, your teams, and your business outcomes, with concrete changes you can implement immediately.

Software Architecture & Systems Engineering

We help design and evaluate software systems that need to scale—applications, APIs, and data platforms—focusing on durability, performance, and long-term maintainability. Whether you’re building new capabilities or untangling existing systems, I bring hands-on expertise to ensure your architecture supports real-world usage, not just ideal-state diagrams.

Cybersecurity Posture & Readiness

Comprehensive security assessments, hardening, and ongoing protection for your digital infrastructure.

Cyber Risk & Exposure Assessment

Navigate the complex landscape of data privacy regulations with confidence.

Who We Are

About Invareo

Invareo is a technology consultancy specializing in web development, systems architecture, cybersecurity, and privacy. We partner with businesses to build, secure, and optimize their technology stack — delivering solutions that are reliable, scalable, and future-proof.

Let's work together

10+

Years Experience

50+

Projects Delivered

100%

Client Focused

24/7

Support Available

Get In Touch

Let's Talk

Have a project in mind or need technology guidance? We'd love to hear from you.

Email Us hello@invareo.com

Location United States

Availability Mon - Fri, 9am - 6pm

1	import axios, { AxiosInstance, AxiosError } from 'axios';
2	import { useAuth } from './hooks/useAuth';
3
4	const MAX_RETRIES = 3;
5	const RETRY_DELAY = 1000;
6
7	const sleep = (ms: number) =>
8	new Promise(r => setTimeout(r, ms));
9
10	const client: AxiosInstance = axios.create({
11	baseURL: process.env.NEXT_PUBLIC_API,
12	withCredentials: true,
13	timeout: 15_000,
14	headers: { 'X-Client': 'invareo-web/1.0' },
15	});
16
17	// Attach bearer token to every request
18	client.interceptors.request.use((cfg) => {
19	const token = localStorage.getItem('access_token');
20	if (token) {
21	cfg.headers.Authorization = `Bearer ${token}`;
22	}
23	return cfg;
24	});
25
26	// Retry with exponential backoff + token refresh
27	client.interceptors.response.use(
28	(res) => res,
29	async (err: AxiosError) => {
30	const cfg = err.config as any;
31	cfg._retry = cfg._retry ?? 0;
32
33	// Refresh token on 401
34	if (err.response?.status === 401) {
35	const { refresh } = useAuth();
36	await refresh();
37	return client.request(cfg);
38	}
39
40	// Retry on 5xx / network errors
41	const status = err.response?.status ?? 0;
42	if ((status >= 500 \|\| status === 0)
43	&& cfg._retry < MAX_RETRIES) {
44	cfg._retry++;
45	const delay = RETRY_DELAY * 2 ** cfg._retry;
46	await sleep(delay);
47	return client.request(cfg);
48	}
49
50	return Promise.reject(err);
51	}
52	);
53
54	export const api = {
55	get: <T>(url: string) =>
56	client.get<T>(url).then(r => r.data),
57	post: <T>(url: string, data?: unknown) =>
58	client.post<T>(url, data).then(r => r.data),
59	del: (url: string) =>
60	client.delete(url).then(r => r.data),
61	};

1	import { useState, useCallback, useEffect } from 'react';
2	import { jwtDecode } from 'jwt-decode';
3
4	interface TokenPayload {
5	sub: string;
6	exp: number;
7	roles: string[];
8	}
9
10	export const useAuth = () => {
11	const [token, setToken] = useState<string \| null>(null);
12	const [user, setUser] = useState<TokenPayload \| null>(null);
13
14	const decode = (t: string): TokenPayload => {
15	const payload = jwtDecode<TokenPayload>(t);
16	if (payload.exp < Date.now() / 1000) {
17	throw new Error('Token expired');
18	}
19	return payload;
20	};
21
22	const login = useCallback(async (
23	email: string, password: string
24	) => {
25	const res = await fetch('/api/auth/login', {
26	method: 'POST',
27	headers: { 'Content-Type': 'application/json' },
28	body: JSON.stringify({ email, password }),
29	credentials: 'include',
30	});
31	if (!res.ok) throw new Error('Invalid credentials');
32	const { access_token } = await res.json();
33	setToken(access_token);
34	setUser(decode(access_token));
35	}, []);
36
37	const logout = useCallback(() => {
38	setToken(null);
39	setUser(null);
40	fetch('/api/auth/logout', {
41	method: 'POST', credentials: 'include'
42	});
43	}, []);
44
45	const hasRole = (role: string): boolean =>
46	user?.roles.includes(role) ?? false;
47
48	useEffect(() => {
49	const saved = localStorage.getItem('access_token');
50	if (saved) {
51	try { setUser(decode(saved)); setToken(saved); }
52	catch { logout(); }
53	}
54	}, [logout]);
55
56	return { token, user, login, logout, hasRole };
57	};

1	import logging, time
2	from functools import wraps
3	from collections import defaultdict
4	from fastapi import Depends, HTTPException, Request
5	from jose import jwt, JWTError
6	from pydantic import BaseModel
7	from app.config import settings
8
9	logger = logging.getLogger("rbac")
10	_rate_map: dict[str, list] = defaultdict(list)
11
12
13	class TokenPayload(BaseModel):
14	sub: str
15	roles: list[str]
16	exp: int
17
18
19	def _check_rate(sub: str, limit: int = 60):
20	now = time.time()
21	window = [t for t in _rate_map[sub] if now - t < 60]
22	_rate_map[sub] = window
23	if len(window) >= limit:
24	raise HTTPException(429, "rate limit exceeded")
25	window.append(now)
26
27
28	def require_roles(
29	*roles: str,
30	rate_limit: int = 60,
31	):
32	"""Dependency: RBAC + rate limiting."""
33	async def _guard(
34	request: Request,
35	token: str = Depends(oauth2_scheme),
36	) -> TokenPayload:
37	try:
38	payload = jwt.decode(
39	token, settings.SECRET_KEY,
40	algorithms=["HS256"],
41	)
42	except JWTError as exc:
43	logger.warning("jwt error: %s", exc)
44	raise HTTPException(401, "invalid token")
45
46	user = TokenPayload(**payload)
47	_check_rate(user.sub, rate_limit)
48
49	if not set(roles) & set(user.roles):
50	logger.warning(
51	"denied %s → %s (has %s)",
52	user.sub, request.url.path, user.roles,
53	)
54	raise HTTPException(403, "insufficient permissions")
55
56	logger.info("allow %s → %s", user.sub, request.url.path)
57	return user
58	return _guard

<b>Hello</b>we're Invareo, a technology consultancy