from . import auth, admin, portal, clients, returns, certificates import socket class MalwareScanError(Exception): pass def clamav_scan_bytes(host: str, port: int, payload: bytes) -> None: """ Uses clamd INSTREAM protocol. Raises MalwareScanError if malware is found or scan fails. """ try: s = socket.create_connection((host, port), timeout=10) s.sendall(b"zINSTREAM\0") chunk_size = 8192 for i in range(0, len(payload), chunk_size): chunk = payload[i:i+chunk_size] s.sendall(len(chunk).to_bytes(4, "big")) s.sendall(chunk) s.sendall((0).to_bytes(4, "big")) resp = s.recv(4096).decode("utf-8", errors="ignore") s.close() except Exception as e: raise MalwareScanError(f"Scan failed: {e}") # Example response: "stream: OK" or "stream: Eicar-Test-Signature FOUND" if "FOUND" in resp.upper(): raise MalwareScanError("Malware detected") if "OK" not in resp.upper(): raise MalwareScanError(f"Unexpected scan response: {resp}") taxhub/ app/ main.py settings.py db.py models.py security.py rbac.py audit.py malware.py integrations/ irs_mef.py state_efile.py bank_products.py tasks/ celery_app.py transmissions.py routers/ auth.py admin.py portal.py clients.py returns.py certificates.py docker-compose.yml requirements.txtfastapi==0.115.6 uvicorn[standard]==0.30.6 SQLAlchemy==2.0.36 psycopg2-binary==2.9.10 pydantic==2.10.3 pydantic-settings==2.6.1 python-jose[cryptography]==3.3.0 passlib[bcrypt]==1.7.4 python-multipart==0.0.20 celery==5.4.0 redis==5.2.0 httpx==0.28.1 services: db: image: postgres:16 environment: POSTGRES_DB: taxhub POSTGRES_USER: taxhub POSTGRES_PASSWORD: taxhub ports: ["5432:5432"] redis: image: redis:7 ports: ["6379:6379"] clamav: image: clamav/clamav:stable ports: ["3310:3310"] from pydantic_settings import BaseSettings class Settings(BaseSettings): APP_NAME: str = "TaxHub" BRAND_NAME: str = "Cross Tax Prep + Bookkeeping" ENV: str = "dev" DATABASE_URL: str = "postgresql+psycopg2://taxhub:taxhub@localhost:5432/taxhub" JWT_SECRET: str = "CHANGE_ME_SUPER_SECRET" JWT_ALG: str = "HS256" JWT_TTL_MINUTES: int = 60 REDIS_URL: str = "redis://localhost:6379/0" # ClamAV (clamd) CLAMAV_HOST: str = "localhost" CLAMAV_PORT: int = 3310 from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, DeclarativeBase from .settings import settings import enum from datetime import datetime from sqlalchemy import ( Column, Integer, String, DateTime, Boolean, Enum, ForeignKey, Text ) from sqlalchemy.orm import relationship from .db import Base class RoleName(str, enum.Enum): admin = "admin" manager = "manager" supervisor = "supervisor" team_lead = "team_lead" mentor = "mentor" tax_associate = "tax_associate" class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True) email = Column(String(255), unique=True, index=True, nullable=False) password_hash = Column(String(255), nullable=False) is_active = Column(Boolean, default=True) role = Column(Enum(RoleName), default=RoleName.tax_associate, nullable=False) created_at = Column(DateTime, default=datetime.utcnow) class Client(Base): __tablename__ = "clients" id = Column(Integer, primary_key=True) full_name = Column(String(255), nullable=False) email = Column(String(255), index=True, nullable=True) phone = Column(String(50), nullable=True) created_at = Column(DateTime, default=datetime.utcnow) returns = relationship("TaxReturn", back_populates="client") class ReturnStatus(str, enum.Enum): draft = "draft" ready_to_transmit = "ready_to_transmit" transmitting = "transmitting" accepted = "accepted" rejected = "rejected" needs_attention = "needs_attention" class TaxReturn(Base): __tablename__ = "tax_returns" id = Column(Integer, primary_key=True) client_id = Column(Integer, ForeignKey("clients.id"), nullable=False) tax_year = Column(Integer, nullable=False) # Identity & compliance tracking fields (store as data; validate in your business rules) ero_name = Column(String(255), nullable=True) ero_efin = Column(String(20), nullable=True) # EFIN preparer_ptin = Column(String(20), nullable=True) # PTIN etin = Column(String(20), nullable=True) # placeholder (if applicable to your workflow) # gating software_purchase_id = Column(String(64), nullable=True) # must exist before transmit bank_product_opt_in = Column(Boolean, default=False) status = Column(Enum(ReturnStatus), default=ReturnStatus.draft, nullable=False) notes = Column(Text, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) # refund tracking fields expected_refund_amount_cents = Column(Integer, nullable=True) refund_status = Column(String(64), nullable=True) # e.g., "pending", "sent", "funded" refund_last_checked_at = Column(DateTime, nullable=True) client = relationship("Client", back_populates="returns") transmissions = relationship("Transmission", back_populates="tax_return") class TransmissionChannel(str, enum.Enum): federal_irs_mef = "federal_irs_mef" state = "state" class Transmission(Base): __tablename__ = "transmissions" id = Column(Integer, primary_key=True) tax_return_id = Column(Integer, ForeignKey("tax_returns.id"), nullable=False) channel = Column(Enum(TransmissionChannel), nullable=False) external_submission_id = Column(String(128), nullable=True) # returned by IRS/state system ack_code = Column(String(64), nullable=True) ack_message = Column(Text, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow) tax_return = relationship("TaxReturn", back_populates="transmissions") class CertificateType(str, enum.Enum): security_training = "security_training" irs_compliance = "irs_compliance" bank_products = "bank_products" bookkeeping = "bookkeeping" class CertificateRecord(Base): __tablename__ = "certificate_records" id = Column(Integer, primary_key=True) user_id = Column(Integer, ForeignKey("users.id"), nullable=False) cert_type = Column(Enum(CertificateType), nullable=False) completed = Column(Boolean, default=False) completed_at = Column(DateTime, nullable=True) evidence_url = Column(String(500), nullable=True) # link to PDF in your portal storage created_at = Column(DateTime, default=datetime.utcnow) class AuditLog(Base): __tablename__ = "audit_logs" id = Column(Integer, primary_key=True) actor_user_id = Column(Integer, nullable=True) action = Column(String(128), nullable=False) entity = Column(String(128), nullable=False) entity_id = Column(String(64), nullable=True) ip = Column(String(64), nullable=True) user_agent = Column(String(255), nullable=True) created_at = Column(DateTime, default=datetime.utcnow) details = Column(Text, nullable=True) # Optional bookkeeping extension hook class LedgerEntry(Base): __tablename__ = "ledger_entries" id = Column(Integer, primary_key=True) client_id = Column(Integer, ForeignKey("clients.id"), nullable=False) entry_date = Column(DateTime, default=datetime.utcnow) description = Column(String(255), nullable=False) amount_cents = Column(Integer, nullable=False) category = Column(String(128), nullable=True) engine = create_engine(settings.DATABASE_URL, pool_pre_ping=True) SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False) class Base(DeclarativeBase): pass def get_db(): db = SessionLocal() try: yield db finally: db.close() settings = Settings() from datetime import datetime, timedelta from jose import jwt from passlib.context import CryptContext from .settings import settings pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def hash_password(plain: str) -> str: return pwd_context.hash(plain) def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) def create_access_token(sub: str) -> str: now = datetime.utcnow() exp = now + timedelta(minutes=settings.JWT_TTL_MINUTES) payload = {"sub": sub, "iat": int(now.timestamp()), "exp": int(exp.timestamp())} return jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.JWT_ALG) def decode_token(token: str) -> dict: return jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALG])

".encode("utf-8") if channel == TransmissionChannel.federal_irs_mef.value: client = IRSMEFClient() result = client.submit_return(payload, ero_efin=tr.ero_efin or "") tx.external_submission_id = result.submission_id tx.ack_code = result.ack_code tx.ack_message = result.ack_message tx.updated_at = datetime.utcnow() # TODO: state client similarly # If submission succeeds, move to ready/awaiting ack; here we keep "transmitting" db.commit() write_audit(db, "TRANSMIT_STARTED", "TaxReturn", str(tr.id), None, details=f"channel={channel}") except Exception as e: tr = db.query(TaxReturn).filter(TaxReturn.id == tax_return_id).first() if tr: tr.status = ReturnStatus.needs_attention db.commit() write_audit(db, "TRANSMIT_ERROR", "TaxReturn", str(tax_return_id), None, details=str(e)) finally: db.close() from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlalchemy.orm import Session from app.db import get_db from app.models import User from app.security import verify_password, create_access_token router = APIRouter(prefix="/auth", tags=["auth"]) class LoginIn(BaseModel): email: str password: str class TokenOut(BaseModel): access_token: str token_type: str = "bearer" @router.post("/login", response_model=TokenOut) def login(data: LoginIn, db: Session = Depends(get_db)): user = db.query(User).filter(User.email == data.email, User.is_active == True).first() if not user or not verify_password(data.password, user.password_hash): raise HTTPException(status_code=401, detail="Invalid credentials") return TokenOut(access_token=create_access_token(sub=user.email)) from fastapi import APIRouter, Depends, Request from pydantic import BaseModel from sqlalchemy.orm import Session from app.db import get_db from app.models import User, RoleName from app.security import hash_password from app.rbac import require_roles from app.audit import write_audit router = APIRouter(prefix="/admin", tags=["admin"]) class UserCreate(BaseModel): email: str password: str role: RoleName @router.post("/users") def create_user( payload: UserCreate, request: Request, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor)), ): user = User(email=payload.email, password_hash=hash_password(payload.password), role=payload.role) db.add(user) db.commit() write_audit(db, "USER_CREATED", "User", str(user.id), None, request.client.host, request.headers.get("user-agent")) return {"id": user.id, "email": user.email, "role": user.role} @router.get("/audit") def list_audit( db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor)), ): rows = db.execute("SELECT id, action, entity, entity_id, created_at FROM audit_logs ORDER BY id DESC LIMIT 200").fetchall() return [{"id": r[0], "action": r[1], "entity": r[2], "entity_id": r[3], "created_at": r[4]} for r in rows] from fastapi import APIRouter, Depends, UploadFile, File, HTTPException from app.rbac import get_current_user from app.settings import settings from app.malware import clamav_scan_bytes, MalwareScanError router = APIRouter(prefix="/portal", tags=["client-portal"]) @router.post("/upload") async def upload_document( file: UploadFile = File(...), user=Depends(get_current_user), ): data = await file.read() if len(data) > 15 * 1024 * 1024: raise HTTPException(status_code=413, detail="File too large") try: clamav_scan_bytes(settings.CLAMAV_HOST, settings.CLAMAV_PORT, data) except MalwareScanError as e: raise HTTPException(status_code=400, detail=f"Upload blocked: {e}") # TODO: store in encrypted object storage (S3 w/ SSE-KMS), generate signed URL, etc. return {"ok": True, "filename": file.filename, "size": len(data)} from fastapi import APIRouter, Depends from pydantic import BaseModel from sqlalchemy.orm import Session from app.db import get_db from app.models import Client from app.rbac import require_roles from app.models import RoleName router = APIRouter(prefix="/clients", tags=["clients"]) class ClientIn(BaseModel): full_name: str email: str | None = None phone: str | None = None @router.post("") def create_client( payload: ClientIn, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor, RoleName.team_lead, RoleName.tax_associate)), ): c = Client(full_name=payload.full_name, email=payload.email, phone=payload.phone) db.add(c) db.commit() return {"id": c.id} from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlalchemy.orm import Session from app.db import get_db from app.models import TaxReturn, ReturnStatus, RoleName, TransmissionChannel from app.rbac import require_roles from app.tasks.transmissions import transmit_return_task router = APIRouter(prefix="/returns", tags=["returns"]) class ReturnCreate(BaseModel): client_id: int tax_year: int ero_name: str | None = None ero_efin: str | None = None preparer_ptin: str | None = None etin: str | None = None bank_product_opt_in: bool = False @router.post("") def create_return( payload: ReturnCreate, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor, RoleName.team_lead, RoleName.tax_associate)), ): tr = TaxReturn( client_id=payload.client_id, tax_year=payload.tax_year, ero_name=payload.ero_name, ero_efin=payload.ero_efin, preparer_ptin=payload.preparer_ptin, etin=payload.etin, bank_product_opt_in=payload.bank_product_opt_in, status=ReturnStatus.draft, ) db.add(tr) db.commit() return {"id": tr.id, "status": tr.status} class PurchaseIn(BaseModel): software_purchase_id: str @router.post("/{return_id}/purchase") def record_purchase( return_id: int, payload: PurchaseIn, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor, RoleName.team_lead)), ): tr = db.query(TaxReturn).filter(TaxReturn.id == return_id).first() if not tr: raise HTTPException(404, "Return not found") tr.software_purchase_id = payload.software_purchase_id if tr.status == ReturnStatus.draft: tr.status = ReturnStatus.ready_to_transmit db.commit() return {"id": tr.id, "status": tr.status} @router.post("/{return_id}/transmit/federal") def transmit_federal( return_id: int, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor, RoleName.team_lead)), ): tr = db.query(TaxReturn).filter(TaxReturn.id == return_id).first() if not tr: raise HTTPException(404, "Return not found") if not tr.software_purchase_id: raise HTTPException(400, "Blocked: software purchase required before filing") transmit_return_task.delay(return_id, TransmissionChannel.federal_irs_mef.value) return {"queued": True, "return_id": return_id} from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlalchemy.orm import Session from app.db import get_db from app.models import TaxReturn, ReturnStatus, RoleName, TransmissionChannel from app.rbac import require_roles from app.tasks.transmissions import transmit_return_task router = APIRouter(prefix="/returns", tags=["returns"]) class ReturnCreate(BaseModel): client_id: int tax_year: int ero_name: str | None = None ero_efin: str | None = None preparer_ptin: str | None = None etin: str | None = None bank_product_opt_in: bool = False @router.post("") def create_return( payload: ReturnCreate, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor, RoleName.team_lead, RoleName.tax_associate)), ): tr = TaxReturn( client_id=payload.client_id, tax_year=payload.tax_year, ero_name=payload.ero_name, ero_efin=payload.ero_efin, preparer_ptin=payload.preparer_ptin, etin=payload.etin, bank_product_opt_in=payload.bank_product_opt_in, status=ReturnStatus.draft, ) db.add(tr) db.commit() return {"id": tr.id, "status": tr.status} class PurchaseIn(BaseModel): software_purchase_id: str @router.post("/{return_id}/purchase") def record_purchase( return_id: int, payload: PurchaseIn, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor, RoleName.team_lead)), ): tr = db.query(TaxReturn).filter(TaxReturn.id == return_id).first() if not tr: raise HTTPException(404, "Return not found") tr.software_purchase_id = payload.software_purchase_id if tr.status == ReturnStatus.draft: tr.status = ReturnStatus.ready_to_transmit db.commit() return {"id": tr.id, "status": tr.status} @router.post("/{return_id}/transmit/federal") def transmit_federal( return_id: int, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor, RoleName.team_lead)), ): tr = db.query(TaxReturn).filter(TaxReturn.id == return_id).first() if not tr: raise HTTPException(404, "Return not found") if not tr.software_purchase_id: raise HTTPException(400, "Blocked: software purchase required before filing") transmit_return_task.delay(return_id, TransmissionChannel.federal_irs_mef.value) return {"queued": True, "return_id": return_id} from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlalchemy.orm import Session from app.db import get_db from app.models import TaxReturn, ReturnStatus, RoleName, TransmissionChannel from app.rbac import require_roles from app.tasks.transmissions import transmit_return_task router = APIRouter(prefix="/returns", tags=["returns"]) class ReturnCreate(BaseModel): client_id: int tax_year: int ero_name: str | None = None ero_efin: str | None = None preparer_ptin: str | None = None etin: str | None = None bank_product_opt_in: bool = False @router.post("") def create_return( payload: ReturnCreate, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor, RoleName.team_lead, RoleName.tax_associate)), ): tr = TaxReturn( client_id=payload.client_id, tax_year=payload.tax_year, ero_name=payload.ero_name, ero_efin=payload.ero_efin, preparer_ptin=payload.preparer_ptin, etin=payload.etin, bank_product_opt_in=payload.bank_product_opt_in, status=ReturnStatus.draft, ) db.add(tr) db.commit() return {"id": tr.id, "status": tr.status} class PurchaseIn(BaseModel): software_purchase_id: str @router.post("/{return_id}/purchase") def record_purchase( return_id: int, payload: PurchaseIn, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor, RoleName.team_lead)), ): tr = db.query(TaxReturn).filter(TaxReturn.id == return_id).first() if not tr: raise HTTPException(404, "Return not found") tr.software_purchase_id = payload.software_purchase_id if tr.status == ReturnStatus.draft: tr.status = ReturnStatus.ready_to_transmit db.commit() return {"id": tr.id, "status": tr.status} @router.post("/{return_id}/transmit/federal") def transmit_federal( return_id: int, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor, RoleName.team_lead)), ): tr = db.query(TaxReturn).filter(TaxReturn.id == return_id).first() if not tr: raise HTTPException(404, "Return not found") if not tr.software_purchase_id: raise HTTPException(400, "Blocked: software purchase required before filing") transmit_return_task.delay(return_id, TransmissionChannel.federal_irs_mef.value) return {"queued": True, "return_id": return_id} from datetime import datetime from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlalchemy.orm import Session from app.db import get_db from app.models import CertificateRecord, CertificateType, RoleName from app.rbac import require_roles router = APIRouter(prefix="/certificates", tags=["certificates"]) class CertCreate(BaseModel): user_id: int cert_type: CertificateType class CertComplete(BaseModel): evidence_url: str | None = None @router.post("") def create_cert( payload: CertCreate, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor, RoleName.mentor)), ): rec = CertificateRecord(user_id=payload.user_id, cert_type=payload.cert_type) db.add(rec) db.commit() return {"id": rec.id} @router.post("/{cert_id}/complete") def complete_cert( cert_id: int, payload: CertComplete, db: Session = Depends(get_db), _=Depends(require_roles(RoleName.admin, RoleName.manager, RoleName.supervisor, RoleName.mentor)), ): rec = db.query(CertificateRecord).filter(CertificateRecord.id == cert_id).first() if not rec: raise HTTPException(404, "Certificate record not found") rec.completed = True rec.completed_at = datetime.utcnow() rec.evidence_url = payload.evidence_url db.commit() return {"id": rec.id, "completed": rec.completed} from fastapi import FastAPI from app.settings import settings from app.db import Base, engine from app.routers import auth, admin, portal, clients, returns, certificates def create_app() -> FastAPI: app = FastAPI(title=settings.APP_NAME) # For demo/dev only. Use Alembic migrations in production. Base.metadata.create_all(bind=engine) app.include_router(auth.router) app.include_router(admin.router) app.include_router(portal.router) app.include_router(clients.router) app.include_router(returns.router) app.include_router(certificates.router) @app.get("/health") def health(): return {"ok": True, "brand": settings.BRAND_NAME} return app app = create_app() celery -A app.tasks.celery_app.celery_app worker -Q transmissions -l info

Becoming a Tax Professional – 1:1 Coaching Session

$159.99 $259.99

Introductory pricing available for a limited time.

Becoming a Tax Professional – 1:1 Coaching is a personalized, one-on-one strategy session designed for individuals who want to enter the tax industry or sharpen their professional foundation.

This session provides direct guidance, real-world insight, and clarity on what it actually takes to operate as a compliant, confident tax professional.

During this session, we cover:

Entry into the tax preparation industry
Required credentials (PTIN, EFIN, software, compliance basics)
Real expectations vs. social-media myths
Common beginner mistakes to avoid
Career paths: preparer, business owner, or expansion strategy
Next-step roadmap tailored to your goals

This is not a group class. This is direct access.

Preparation, licensing, or filing services are not included and are quoted separately if needed.

Program Level: